added new module 'chat.py'

chatmastermind/chat.py

@@ -0,0 +1,204 @@
+"""
+Module implementing various chat classes and functions for managing a chat history.
+"""
+import shutil
+from pprint import PrettyPrinter
+import pathlib
+from dataclasses import dataclass, field
+from typing import TypeVar, Type, Optional, ClassVar, Any
+from .message import Message, MessageFilter, MessageError
+
+ChatInst = TypeVar('ChatInst', bound='Chat')
+ChatDBInst = TypeVar('ChatDBInst', bound='ChatDB')
+
+
+class ChatError(Exception):
+    pass
+
+
+def terminal_width() -> int:
+    return shutil.get_terminal_size().columns
+
+
+def pp(*args: Any, **kwargs: Any) -> None:
+    return PrettyPrinter(width=terminal_width()).pprint(*args, **kwargs)
+
+
+@dataclass
+class Chat:
+    """
+    A class containing a complete chat history.
+    """
+
+    messages: list[Message]
+
+    def filter(self, mfilter: MessageFilter) -> None:
+        """
+        Use 'Message.match(mfilter) to remove all messages that
+        don't fulfill the filter requirements.
+        """
+        self.messages = [m for m in self.messages if m.match(mfilter)]
+
+    def sort(self, reverse: bool = False) -> None:
+        """
+        Sort the messages according to 'Message.msg_id()'.
+        """
+        try:
+            # the message may not have an ID if it doesn't have a file_path
+            self.messages.sort(key=lambda m: m.msg_id(), reverse=reverse)
+        except MessageError:
+            pass
+
+    def clear(self) -> None:
+        """
+        Delete all messages.
+        """
+        self.messages = []
+
+    def add_msgs(self, msgs: list[Message]) -> None:
+        """
+        Add new messages and sort them if possible.
+        """
+        self.messages += msgs
+        self.sort()
+
+    def print(self, dump: bool = False) -> None:
+        if dump:
+            pp(self)
+            return
+        # for message in self.messages:
+        #     text_too_long = len(message['content']) > terminal_width() - len(message['role']) - 2
+        #     if source_code:
+        #         display_source_code(message['content'])
+        #         continue
+        #     if message['role'] == 'user':
+        #         print('-' * terminal_width())
+        #     if text_too_long:
+        #         print(f"{message['role'].upper()}:")
+        #         print(message['content'])
+        #     else:
+        #         print(f"{message['role'].upper()}: {message['content']}")
+
+
+@dataclass
+class ChatDB(Chat):
+    """
+    A 'Chat' class that is bound to a given directory structure. Supports reading
+    and writing messages from / to that structure. Such a structure consists of
+    two directories: a 'cache directory', where all messages are temporarily
+    stored, and a 'DB' directory, where selected messages can be stored
+    persistently.
+    """
+
+    default_file_suffix: ClassVar[str] = '.txt'
+
+    cache_path: pathlib.Path
+    db_path: pathlib.Path
+    # a MessageFilter that all messages must match (if given)
+    mfilter: Optional[MessageFilter] = None
+    file_suffix: str = default_file_suffix
+    # the glob pattern for all messages
+    glob: Optional[str] = None
+    # set containing all file names of the current messages
+    message_files: set[str] = field(default_factory=set, repr=False)
+
+    @classmethod
+    def from_dir(cls: Type[ChatDBInst],
+                 cache_path: pathlib.Path,
+                 db_path: pathlib.Path,
+                 glob: Optional[str] = None,
+                 mfilter: Optional[MessageFilter] = None) -> ChatDBInst:
+        """
+        Create a 'ChatDB' instance from the given directory structure.
+        Reads all messages from 'db_path' into the local message list.
+        Parameters:
+        * 'cache_path': path to the directory for temporary messages
+        * 'db_path': path to the directory for persistent messages
+        * 'glob' fs specified, files will be filtered using 'path.glob()',
+          otherwise it uses 'path.iterdir()'.
+        * 'mfilter': use with 'Message.from_file()' to filter messages
+          when reading them.
+        """
+        messages: list[Message] = []
+        message_files: set[str] = set()
+        file_iter = db_path.glob(glob) if glob else db_path.iterdir()
+        for file_path in sorted(file_iter):
+            if file_path.is_file():
+                try:
+                    message = Message.from_file(file_path, mfilter)
+                    if message:
+                        messages.append(message)
+                        message_files.add(file_path.name)
+                except MessageError as e:
+                    print(f"Error processing message in '{file_path}': {str(e)}")
+        return cls(messages, cache_path, db_path, mfilter,
+                   cls.default_file_suffix, glob, message_files)
+
+    @classmethod
+    def from_messages(cls: Type[ChatDBInst],
+                      cache_path: pathlib.Path,
+                      db_path: pathlib.Path,
+                      messages: list[Message],
+                      mfilter: Optional[MessageFilter]) -> ChatDBInst:
+        """
+        Create a ChatDB instance from the given message list. Note that the next
+        call to 'dump()' will write all files in order to synchronize the messages.
+        Similarly, 'update()' will read all messages, so you may end up with a lot
+        of duplicates when using 'update()' first.
+        """
+        return cls(messages, cache_path, db_path, mfilter)
+
+    def get_next_fid(self) -> int:
+        next_fname = self.db_path / '.next'
+        try:
+            with open(next_fname, 'r') as f:
+                return int(f.read()) + 1
+        except Exception:
+            return 1
+
+    def set_next_fid(self, fid: int) -> None:
+        next_fname = self.db_path / '.next'
+        with open(next_fname, 'w') as f:
+            f.write(f'{fid}')
+
+    def dump(self, to_db: bool = False, force_all: bool = False) -> None:
+        """
+        Write all messages to 'cache_path' (or 'db_path' if 'to_db' is True). If a message
+        has no file_path, a new one will be created. By default, only messages that have
+        not been written (or read) before will be dumped. Use 'force_all' to force writing
+        all message files.
+        """
+        for message in self.messages:
+            # skip messages that we have already written (or read)
+            if message.file_path and message.file_path in self.message_files and not force_all:
+                continue
+            file_path = message.file_path
+            if not file_path:
+                fid = self.get_next_fid()
+                fname = f"{fid:04d}{self.file_suffix}"
+                file_path = self.db_path / fname if to_db else self.cache_path / fname
+                self.set_next_fid(fid)
+            message.to_file(file_path)
+
+    def update(self, from_cache: bool = False, force_all: bool = False) -> None:
+        """
+        Read new messages from 'db_path' (or 'cache_path' if 'from_cache' is true).
+        By default, only messages that have not been read (or written) before will
+        be read. Use 'force_all' to force reading all messages.
+        """
+        if from_cache:
+            file_iter = self.cache_path.glob(self.glob) if self.glob else self.cache_path.iterdir()
+        else:
+            file_iter = self.cache_path.glob(self.glob) if self.glob else self.cache_path.iterdir()
+        for file_path in sorted(file_iter):
+            if file_path.is_file():
+                if file_path.name in self.message_files and not force_all:
+                    continue
+                try:
+                    message = Message.from_file(file_path, self.mfilter)
+                    if message:
+                        self.messages.append(message)
+                        self.message_files.add(file_path.name)
+                except MessageError as e:
+                    print(f"Error processing message in '{file_path}': {str(e)}")
+            self.sort()

tests/test_message.py

@@ -1,5 +1,9 @@
+import pathlib
+import tempfile
+from typing import cast
 from .test_main import CmmTestCase
-from chatmastermind.message import source_code, MessageError, Question, Answer
+from chatmastermind.message import source_code, Message, MessageError, Question, Answer, AILine, ModelLine, MessageFilter
+from chatmastermind.tags import Tag, TagLine
 
 
 class SourceCodeTestCase(CmmTestCase):
@@ -76,3 +80,539 @@ class AnswerTestCase(CmmTestCase):
         answer = Answer("No")
         self.assertIsInstance(answer, Answer)
         self.assertEqual(answer, "No")
+
+
+class MessageToFileTxtTestCase(CmmTestCase):
+    def setUp(self) -> None:
+        self.file = tempfile.NamedTemporaryFile(delete=False, suffix='.txt')
+        self.file_path = pathlib.Path(self.file.name)
+        self.message_complete = Message(Question('This is a question.'),
+                                        Answer('This is an answer.'),
+                                        {Tag('tag1'), Tag('tag2')},
+                                        ai='ChatGPT',
+                                        model='gpt-3.5-turbo',
+                                        file_path=self.file_path)
+        self.message_min = Message(Question('This is a question.'),
+                                   file_path=self.file_path)
+
+    def tearDown(self) -> None:
+        self.file.close()
+        self.file_path.unlink()
+
+    def test_to_file_txt_complete(self) -> None:
+        self.message_complete.to_file(self.file_path)
+
+        with open(self.file_path, "r") as fd:
+            content = fd.read()
+        expected_content = f"""{TagLine.prefix} tag1 tag2
+{AILine.prefix} ChatGPT
+{ModelLine.prefix} gpt-3.5-turbo
+{Question.txt_header}
+This is a question.
+{Answer.txt_header}
+This is an answer.
+"""
+        self.assertEqual(content, expected_content)
+
+    def test_to_file_txt_min(self) -> None:
+        self.message_min.to_file(self.file_path)
+
+        with open(self.file_path, "r") as fd:
+            content = fd.read()
+        expected_content = f"""{Question.txt_header}
+This is a question.
+"""
+        self.assertEqual(content, expected_content)
+
+    def test_to_file_unsupported_file_type(self) -> None:
+        unsupported_file_path = pathlib.Path("example.doc")
+        with self.assertRaises(MessageError) as cm:
+            self.message_complete.to_file(unsupported_file_path)
+        self.assertEqual(str(cm.exception), "File type '.doc' is not supported")
+
+    def test_to_file_no_file_path(self) -> None:
+        """
+        Provoke an exception using an empty path.
+        """
+        with self.assertRaises(MessageError) as cm:
+            # clear the internal file_path
+            self.message_complete.file_path = None
+            self.message_complete.to_file(None)
+        self.assertEqual(str(cm.exception), "Got no valid path to write message")
+        # reset the internal file_path
+        self.message_complete.file_path = self.file_path
+
+
+class MessageToFileYamlTestCase(CmmTestCase):
+    def setUp(self) -> None:
+        self.file = tempfile.NamedTemporaryFile(delete=False, suffix='.yaml')
+        self.file_path = pathlib.Path(self.file.name)
+        self.message_complete = Message(Question('This is a question.'),
+                                        Answer('This is an answer.'),
+                                        {Tag('tag1'), Tag('tag2')},
+                                        ai='ChatGPT',
+                                        model='gpt-3.5-turbo',
+                                        file_path=self.file_path)
+        self.message_multiline = Message(Question('This is a\nmultiline question.'),
+                                         Answer('This is a\nmultiline answer.'),
+                                         {Tag('tag1'), Tag('tag2')},
+                                         ai='ChatGPT',
+                                         model='gpt-3.5-turbo',
+                                         file_path=self.file_path)
+        self.message_min = Message(Question('This is a question.'),
+                                   file_path=self.file_path)
+
+    def tearDown(self) -> None:
+        self.file.close()
+        self.file_path.unlink()
+
+    def test_to_file_yaml_complete(self) -> None:
+        self.message_complete.to_file(self.file_path)
+
+        with open(self.file_path, "r") as fd:
+            content = fd.read()
+        expected_content = f"""{Question.yaml_key}: This is a question.
+{Answer.yaml_key}: This is an answer.
+{Message.ai_yaml_key}: ChatGPT
+{Message.model_yaml_key}: gpt-3.5-turbo
+{Message.tags_yaml_key}:
+- tag1
+- tag2
+"""
+        self.assertEqual(content, expected_content)
+
+    def test_to_file_yaml_multiline(self) -> None:
+        self.message_multiline.to_file(self.file_path)
+
+        with open(self.file_path, "r") as fd:
+            content = fd.read()
+        expected_content = f"""{Question.yaml_key}: |-
+  This is a
+  multiline question.
+{Answer.yaml_key}: |-
+  This is a
+  multiline answer.
+{Message.ai_yaml_key}: ChatGPT
+{Message.model_yaml_key}: gpt-3.5-turbo
+{Message.tags_yaml_key}:
+- tag1
+- tag2
+"""
+        self.assertEqual(content, expected_content)
+
+    def test_to_file_yaml_min(self) -> None:
+        self.message_min.to_file(self.file_path)
+
+        with open(self.file_path, "r") as fd:
+            content = fd.read()
+        expected_content = f"{Question.yaml_key}: This is a question.\n"
+        self.assertEqual(content, expected_content)
+
+
+class MessageFromFileTxtTestCase(CmmTestCase):
+    def setUp(self) -> None:
+        self.file = tempfile.NamedTemporaryFile(delete=False, suffix='.txt')
+        self.file_path = pathlib.Path(self.file.name)
+        with open(self.file_path, "w") as fd:
+            fd.write(f"""{TagLine.prefix} tag1 tag2
+{AILine.prefix} ChatGPT
+{ModelLine.prefix} gpt-3.5-turbo
+{Question.txt_header}
+This is a question.
+{Answer.txt_header}
+This is an answer.
+""")
+        self.file_min = tempfile.NamedTemporaryFile(delete=False, suffix='.txt')
+        self.file_path_min = pathlib.Path(self.file_min.name)
+        with open(self.file_path_min, "w") as fd:
+            fd.write(f"""{Question.txt_header}
+This is a question.
+""")
+
+    def tearDown(self) -> None:
+        self.file.close()
+        self.file_min.close()
+        self.file_path.unlink()
+        self.file_path_min.unlink()
+
+    def test_from_file_txt_complete(self) -> None:
+        """
+        Read a complete message (with all optional values).
+        """
+        message = Message.from_file(self.file_path)
+        self.assertIsNotNone(message)
+        self.assertIsInstance(message, Message)
+        if message:  # mypy bug
+            self.assertEqual(message.question, 'This is a question.')
+            self.assertEqual(message.answer, 'This is an answer.')
+            self.assertSetEqual(cast(set[Tag], message.tags), {Tag('tag1'), Tag('tag2')})
+            self.assertEqual(message.ai, 'ChatGPT')
+            self.assertEqual(message.model, 'gpt-3.5-turbo')
+            self.assertEqual(message.file_path, self.file_path)
+
+    def test_from_file_txt_min(self) -> None:
+        """
+        Read a message with only required values.
+        """
+        message = Message.from_file(self.file_path_min)
+        self.assertIsNotNone(message)
+        self.assertIsInstance(message, Message)
+        if message:  # mypy bug
+            self.assertEqual(message.question, 'This is a question.')
+            self.assertEqual(message.file_path, self.file_path_min)
+            self.assertIsNone(message.answer)
+
+    def test_from_file_txt_tags_match(self) -> None:
+        message = Message.from_file(self.file_path,
+                                    MessageFilter(tags_or={Tag('tag1')}))
+        self.assertIsNotNone(message)
+        self.assertIsInstance(message, Message)
+        if message:  # mypy bug
+            self.assertEqual(message.question, 'This is a question.')
+            self.assertEqual(message.answer, 'This is an answer.')
+            self.assertSetEqual(cast(set[Tag], message.tags), {Tag('tag1'), Tag('tag2')})
+            self.assertEqual(message.file_path, self.file_path)
+
+    def test_from_file_txt_tags_dont_match(self) -> None:
+        message = Message.from_file(self.file_path,
+                                    MessageFilter(tags_or={Tag('tag3')}))
+        self.assertIsNone(message)
+
+    def test_from_file_txt_no_tags_dont_match(self) -> None:
+        message = Message.from_file(self.file_path_min,
+                                    MessageFilter(tags_or={Tag('tag1')}))
+        self.assertIsNone(message)
+
+    def test_from_file_txt_no_tags_match_tags_not(self) -> None:
+        message = Message.from_file(self.file_path_min,
+                                    MessageFilter(tags_not={Tag('tag1')}))
+        self.assertIsNotNone(message)
+        self.assertIsInstance(message, Message)
+        if message:  # mypy bug
+            self.assertEqual(message.question, 'This is a question.')
+            self.assertSetEqual(cast(set[Tag], message.tags), set())
+            self.assertEqual(message.file_path, self.file_path_min)
+
+    def test_from_file_not_exists(self) -> None:
+        file_not_exists = pathlib.Path("example.txt")
+        with self.assertRaises(MessageError) as cm:
+            Message.from_file(file_not_exists)
+        self.assertEqual(str(cm.exception), f"Message file '{file_not_exists}' does not exist")
+
+    def test_from_file_txt_question_match(self) -> None:
+        message = Message.from_file(self.file_path,
+                                    MessageFilter(question_contains='question'))
+        self.assertIsNotNone(message)
+        self.assertIsInstance(message, Message)
+
+    def test_from_file_txt_answer_match(self) -> None:
+        message = Message.from_file(self.file_path,
+                                    MessageFilter(answer_contains='answer'))
+        self.assertIsNotNone(message)
+        self.assertIsInstance(message, Message)
+
+    def test_from_file_txt_answer_available(self) -> None:
+        message = Message.from_file(self.file_path,
+                                    MessageFilter(answer_state='available'))
+        self.assertIsNotNone(message)
+        self.assertIsInstance(message, Message)
+
+    def test_from_file_txt_answer_missing(self) -> None:
+        message = Message.from_file(self.file_path_min,
+                                    MessageFilter(answer_state='missing'))
+        self.assertIsNotNone(message)
+        self.assertIsInstance(message, Message)
+
+    def test_from_file_txt_question_doesnt_match(self) -> None:
+        message = Message.from_file(self.file_path,
+                                    MessageFilter(question_contains='answer'))
+        self.assertIsNone(message)
+
+    def test_from_file_txt_answer_doesnt_match(self) -> None:
+        message = Message.from_file(self.file_path,
+                                    MessageFilter(answer_contains='question'))
+        self.assertIsNone(message)
+
+    def test_from_file_txt_answer_not_exists(self) -> None:
+        message = Message.from_file(self.file_path_min,
+                                    MessageFilter(answer_contains='answer'))
+        self.assertIsNone(message)
+
+    def test_from_file_txt_answer_not_available(self) -> None:
+        message = Message.from_file(self.file_path_min,
+                                    MessageFilter(answer_state='available'))
+        self.assertIsNone(message)
+
+    def test_from_file_txt_answer_not_missing(self) -> None:
+        message = Message.from_file(self.file_path,
+                                    MessageFilter(answer_state='missing'))
+        self.assertIsNone(message)
+
+    def test_from_file_txt_ai_match(self) -> None:
+        message = Message.from_file(self.file_path,
+                                    MessageFilter(ai='ChatGPT'))
+        self.assertIsNotNone(message)
+        self.assertIsInstance(message, Message)
+
+    def test_from_file_txt_ai_doesnt_match(self) -> None:
+        message = Message.from_file(self.file_path,
+                                    MessageFilter(ai='Foo'))
+        self.assertIsNone(message)
+
+    def test_from_file_txt_model_match(self) -> None:
+        message = Message.from_file(self.file_path,
+                                    MessageFilter(model='gpt-3.5-turbo'))
+        self.assertIsNotNone(message)
+        self.assertIsInstance(message, Message)
+
+    def test_from_file_txt_model_doesnt_match(self) -> None:
+        message = Message.from_file(self.file_path,
+                                    MessageFilter(model='Bar'))
+        self.assertIsNone(message)
+
+
+class MessageFromFileYamlTestCase(CmmTestCase):
+    def setUp(self) -> None:
+        self.file = tempfile.NamedTemporaryFile(delete=False, suffix='.yaml')
+        self.file_path = pathlib.Path(self.file.name)
+        with open(self.file_path, "w") as fd:
+            fd.write(f"""
+{Question.yaml_key}: |-
+  This is a question.
+{Answer.yaml_key}: |-
+  This is an answer.
+{Message.ai_yaml_key}: ChatGPT
+{Message.model_yaml_key}: gpt-3.5-turbo
+{Message.tags_yaml_key}:
+  - tag1
+  - tag2
+""")
+        self.file_min = tempfile.NamedTemporaryFile(delete=False, suffix='.yaml')
+        self.file_path_min = pathlib.Path(self.file_min.name)
+        with open(self.file_path_min, "w") as fd:
+            fd.write(f"""
+{Question.yaml_key}: |-
+  This is a question.
+""")
+
+    def tearDown(self) -> None:
+        self.file.close()
+        self.file_path.unlink()
+        self.file_min.close()
+        self.file_path_min.unlink()
+
+    def test_from_file_yaml_complete(self) -> None:
+        """
+        Read a complete message (with all optional values).
+        """
+        message = Message.from_file(self.file_path)
+        self.assertIsInstance(message, Message)
+        self.assertIsNotNone(message)
+        if message:  # mypy bug
+            self.assertEqual(message.question, 'This is a question.')
+            self.assertEqual(message.answer, 'This is an answer.')
+            self.assertSetEqual(cast(set[Tag], message.tags), {Tag('tag1'), Tag('tag2')})
+            self.assertEqual(message.ai, 'ChatGPT')
+            self.assertEqual(message.model, 'gpt-3.5-turbo')
+            self.assertEqual(message.file_path, self.file_path)
+
+    def test_from_file_yaml_min(self) -> None:
+        """
+        Read a message with only the required values.
+        """
+        message = Message.from_file(self.file_path_min)
+        self.assertIsInstance(message, Message)
+        self.assertIsNotNone(message)
+        if message:  # mypy bug
+            self.assertEqual(message.question, 'This is a question.')
+            self.assertSetEqual(cast(set[Tag], message.tags), set())
+            self.assertEqual(message.file_path, self.file_path_min)
+            self.assertIsNone(message.answer)
+
+    def test_from_file_not_exists(self) -> None:
+        file_not_exists = pathlib.Path("example.yaml")
+        with self.assertRaises(MessageError) as cm:
+            Message.from_file(file_not_exists)
+        self.assertEqual(str(cm.exception), f"Message file '{file_not_exists}' does not exist")
+
+    def test_from_file_yaml_tags_match(self) -> None:
+        message = Message.from_file(self.file_path,
+                                    MessageFilter(tags_or={Tag('tag1')}))
+        self.assertIsNotNone(message)
+        self.assertIsInstance(message, Message)
+        if message:  # mypy bug
+            self.assertEqual(message.question, 'This is a question.')
+            self.assertEqual(message.answer, 'This is an answer.')
+            self.assertSetEqual(cast(set[Tag], message.tags), {Tag('tag1'), Tag('tag2')})
+            self.assertEqual(message.file_path, self.file_path)
+
+    def test_from_file_yaml_tags_dont_match(self) -> None:
+        message = Message.from_file(self.file_path,
+                                    MessageFilter(tags_or={Tag('tag3')}))
+        self.assertIsNone(message)
+
+    def test_from_file_yaml_no_tags_dont_match(self) -> None:
+        message = Message.from_file(self.file_path_min,
+                                    MessageFilter(tags_or={Tag('tag1')}))
+        self.assertIsNone(message)
+
+    def test_from_file_yaml_no_tags_match_tags_not(self) -> None:
+        message = Message.from_file(self.file_path_min,
+                                    MessageFilter(tags_not={Tag('tag1')}))
+        self.assertIsNotNone(message)
+        self.assertIsInstance(message, Message)
+        if message:  # mypy bug
+            self.assertEqual(message.question, 'This is a question.')
+            self.assertSetEqual(cast(set[Tag], message.tags), set())
+            self.assertEqual(message.file_path, self.file_path_min)
+
+    def test_from_file_yaml_question_match(self) -> None:
+        message = Message.from_file(self.file_path,
+                                    MessageFilter(question_contains='question'))
+        self.assertIsNotNone(message)
+        self.assertIsInstance(message, Message)
+
+    def test_from_file_yaml_answer_match(self) -> None:
+        message = Message.from_file(self.file_path,
+                                    MessageFilter(answer_contains='answer'))
+        self.assertIsNotNone(message)
+        self.assertIsInstance(message, Message)
+
+    def test_from_file_yaml_answer_available(self) -> None:
+        message = Message.from_file(self.file_path,
+                                    MessageFilter(answer_state='available'))
+        self.assertIsNotNone(message)
+        self.assertIsInstance(message, Message)
+
+    def test_from_file_yaml_answer_missing(self) -> None:
+        message = Message.from_file(self.file_path_min,
+                                    MessageFilter(answer_state='missing'))
+        self.assertIsNotNone(message)
+        self.assertIsInstance(message, Message)
+
+    def test_from_file_yaml_question_doesnt_match(self) -> None:
+        message = Message.from_file(self.file_path,
+                                    MessageFilter(question_contains='answer'))
+        self.assertIsNone(message)
+
+    def test_from_file_yaml_answer_doesnt_match(self) -> None:
+        message = Message.from_file(self.file_path,
+                                    MessageFilter(answer_contains='question'))
+        self.assertIsNone(message)
+
+    def test_from_file_yaml_answer_not_exists(self) -> None:
+        message = Message.from_file(self.file_path_min,
+                                    MessageFilter(answer_contains='answer'))
+        self.assertIsNone(message)
+
+    def test_from_file_yaml_answer_not_available(self) -> None:
+        message = Message.from_file(self.file_path_min,
+                                    MessageFilter(answer_state='available'))
+        self.assertIsNone(message)
+
+    def test_from_file_yaml_answer_not_missing(self) -> None:
+        message = Message.from_file(self.file_path,
+                                    MessageFilter(answer_state='missing'))
+        self.assertIsNone(message)
+
+    def test_from_file_yaml_ai_match(self) -> None:
+        message = Message.from_file(self.file_path,
+                                    MessageFilter(ai='ChatGPT'))
+        self.assertIsNotNone(message)
+        self.assertIsInstance(message, Message)
+
+    def test_from_file_yaml_ai_doesnt_match(self) -> None:
+        message = Message.from_file(self.file_path,
+                                    MessageFilter(ai='Foo'))
+        self.assertIsNone(message)
+
+    def test_from_file_yaml_model_match(self) -> None:
+        message = Message.from_file(self.file_path,
+                                    MessageFilter(model='gpt-3.5-turbo'))
+        self.assertIsNotNone(message)
+        self.assertIsInstance(message, Message)
+
+    def test_from_file_yaml_model_doesnt_match(self) -> None:
+        message = Message.from_file(self.file_path,
+                                    MessageFilter(model='Bar'))
+        self.assertIsNone(message)
+
+
+class TagsFromFileTestCase(CmmTestCase):
+    def setUp(self) -> None:
+        self.file_txt = tempfile.NamedTemporaryFile(delete=False, suffix='.txt')
+        self.file_path_txt = pathlib.Path(self.file_txt.name)
+        with open(self.file_path_txt, "w") as fd:
+            fd.write(f"""{TagLine.prefix} tag1 tag2
+{Question.txt_header}
+This is a question.
+{Answer.txt_header}
+This is an answer.
+""")
+        self.file_yaml = tempfile.NamedTemporaryFile(delete=False, suffix='.yaml')
+        self.file_path_yaml = pathlib.Path(self.file_yaml.name)
+        with open(self.file_path_yaml, "w") as fd:
+            fd.write(f"""
+{Question.yaml_key}: |-
+  This is a question.
+{Answer.yaml_key}: |-
+  This is an answer.
+{Message.tags_yaml_key}:
+  - tag1
+  - tag2
+""")
+
+    def tearDown(self) -> None:
+        self.file_txt.close()
+        self.file_path_txt.unlink()
+        self.file_yaml.close()
+        self.file_path_yaml.unlink()
+
+    def test_tags_from_file_txt(self) -> None:
+        tags = Message.tags_from_file(self.file_path_txt)
+        self.assertSetEqual(tags, {Tag('tag1'), Tag('tag2')})
+
+    def test_tags_from_file_yaml(self) -> None:
+        tags = Message.tags_from_file(self.file_path_yaml)
+        self.assertSetEqual(tags, {Tag('tag1'), Tag('tag2')})
+
+
+class MessageIDTestCase(CmmTestCase):
+    def setUp(self) -> None:
+        self.file = tempfile.NamedTemporaryFile(delete=False, suffix='.txt')
+        self.file_path = pathlib.Path(self.file.name)
+        self.message = Message(Question('This is a question.'),
+                               file_path=self.file_path)
+        self.message_no_file_path = Message(Question('This is a question.'))
+
+    def tearDown(self) -> None:
+        self.file.close()
+        self.file_path.unlink()
+
+    def test_msg_id_txt(self) -> None:
+        self.assertEqual(self.message.msg_id(), self.file_path.name)
+
+    def test_msg_id_txt_exception(self) -> None:
+        with self.assertRaises(MessageError):
+            self.message_no_file_path.msg_id()
+
+
+class MessageHashTestCase(CmmTestCase):
+    def setUp(self) -> None:
+        self.message1 = Message(Question('This is a question.'),
+                                file_path=pathlib.Path('/tmp/foo/bla'))
+        self.message2 = Message(Question('This is a new question.'),
+                                file_path=pathlib.Path('/tmp/foo/bla'))
+        self.message3 = Message(Question('This is a question.'),
+                                Answer('This is an answer.'),
+                                file_path=pathlib.Path('/tmp/foo/bla'))
+        # message4 is a copy of message1, because 'file_path'
+        # is ignored for hashing and comparison
+        self.message4 = Message(Question('This is a question.'),
+                                file_path=pathlib.Path('foobla'))
+
+    def test_set_hashing(self) -> None:
+        msgs: set[Message] = {self.message1, self.message2, self.message3, self.message4}
+        self.assertEqual(len(msgs), 3)
+        for msg in [self.message1, self.message2, self.message3]:
+            self.assertIn(msg, msgs)