Compare commits
7 Commits
5e818d4af7
...
4c674b80a6
| Author | SHA1 | Date | |
|---|---|---|---|
| 4c674b80a6 | |||
| a3d3ee1c93 | |||
| 0a83aa2c6f | |||
| 8e14ff0c16 | |||
| 12d460a6d4 | |||
| 09a082f369 | |||
| 99885ba1e1 |
@ -3,19 +3,32 @@ Module implementing message related functions and classes.
|
|||||||
"""
|
"""
|
||||||
import pathlib
|
import pathlib
|
||||||
import yaml
|
import yaml
|
||||||
from typing import Type, TypeVar, ClassVar, Optional, Any
|
from typing import Type, TypeVar, ClassVar, Optional, Any, Union
|
||||||
from dataclasses import dataclass, asdict
|
from dataclasses import dataclass, asdict
|
||||||
from .tags import Tag, TagLine
|
from .tags import Tag, TagLine
|
||||||
|
|
||||||
QuestionInst = TypeVar('QuestionInst', bound='Question')
|
QuestionInst = TypeVar('QuestionInst', bound='Question')
|
||||||
AnswerInst = TypeVar('AnswerInst', bound='Answer')
|
AnswerInst = TypeVar('AnswerInst', bound='Answer')
|
||||||
MessageInst = TypeVar('MessageInst', bound='Message')
|
MessageInst = TypeVar('MessageInst', bound='Message')
|
||||||
|
YamlDict = dict[str, Union[QuestionInst, AnswerInst, set[Tag]]]
|
||||||
|
|
||||||
|
|
||||||
class MessageError(Exception):
|
class MessageError(Exception):
|
||||||
pass
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
def str_presenter(dumper: yaml.Dumper, data: str) -> yaml.ScalarNode:
|
||||||
|
"""
|
||||||
|
Changes the YAML dump style to multiline syntax for multiline strings.
|
||||||
|
"""
|
||||||
|
if len(data.splitlines()) > 1:
|
||||||
|
return dumper.represent_scalar('tag:yaml.org,2002:str', data, style='|')
|
||||||
|
return dumper.represent_scalar('tag:yaml.org,2002:str', data)
|
||||||
|
|
||||||
|
|
||||||
|
yaml.add_representer(str, str_presenter)
|
||||||
|
|
||||||
|
|
||||||
def source_code(text: str, include_delims: bool = False) -> list[str]:
|
def source_code(text: str, include_delims: bool = False) -> list[str]:
|
||||||
"""
|
"""
|
||||||
Extract all source code sections from the given text, i. e. all lines
|
Extract all source code sections from the given text, i. e. all lines
|
||||||
@ -154,7 +167,9 @@ class Message():
|
|||||||
* Question
|
* Question
|
||||||
* Answer.Header
|
* Answer.Header
|
||||||
For '.yaml':
|
For '.yaml':
|
||||||
TODO
|
* question: single or multiline string
|
||||||
|
* answer: single or multiline string
|
||||||
|
* tags: list of strings
|
||||||
"""
|
"""
|
||||||
if not file_path.exists():
|
if not file_path.exists():
|
||||||
raise MessageError(f"Message file '{file_path}' does not exist")
|
raise MessageError(f"Message file '{file_path}' does not exist")
|
||||||
@ -175,7 +190,6 @@ class Message():
|
|||||||
return cls(question, answer, tags, file_path)
|
return cls(question, answer, tags, file_path)
|
||||||
else: # '.yaml'
|
else: # '.yaml'
|
||||||
with open(file_path, "r") as fd:
|
with open(file_path, "r") as fd:
|
||||||
# FIXME: use the actual YAML format
|
|
||||||
data = yaml.load(fd, Loader=yaml.FullLoader)
|
data = yaml.load(fd, Loader=yaml.FullLoader)
|
||||||
data['file_path'] = file_path
|
data['file_path'] = file_path
|
||||||
return cls.from_dict(data)
|
return cls.from_dict(data)
|
||||||
@ -190,7 +204,9 @@ class Message():
|
|||||||
* Answer.Header
|
* Answer.Header
|
||||||
* Answer
|
* Answer
|
||||||
For '.yaml':
|
For '.yaml':
|
||||||
TODO
|
* question: single or multiline string
|
||||||
|
* answer: single or multiline string
|
||||||
|
* tags: list of strings
|
||||||
"""
|
"""
|
||||||
if file_path:
|
if file_path:
|
||||||
self.file_path = file_path
|
self.file_path = file_path
|
||||||
@ -204,7 +220,14 @@ class Message():
|
|||||||
fd.write(f'{TagLine.from_set(msg_tags)}\n')
|
fd.write(f'{TagLine.from_set(msg_tags)}\n')
|
||||||
fd.write(f'{Question.header}\n{self.question}\n')
|
fd.write(f'{Question.header}\n{self.question}\n')
|
||||||
fd.write(f'{Answer.header}\n{self.answer}\n')
|
fd.write(f'{Answer.header}\n{self.answer}\n')
|
||||||
# FIXME: write YAML format
|
elif self.file_path.suffix == '.yaml':
|
||||||
|
with open(self.file_path, "w") as fd:
|
||||||
|
data: YamlDict = {'question': str(self.question)}
|
||||||
|
if self.answer:
|
||||||
|
data['answer'] = str(self.answer)
|
||||||
|
if self.tags:
|
||||||
|
data['tags'] = sorted([str(tag) for tag in self.tags])
|
||||||
|
yaml.dump(data, fd)
|
||||||
|
|
||||||
def as_dict(self) -> dict[str, Any]:
|
def as_dict(self) -> dict[str, Any]:
|
||||||
return asdict(self)
|
return asdict(self)
|
||||||
|
|||||||
@ -86,8 +86,8 @@ class MessageToFileTxtTestCase(CmmTestCase):
|
|||||||
def setUp(self) -> None:
|
def setUp(self) -> None:
|
||||||
self.file = tempfile.NamedTemporaryFile(delete=False, suffix='.txt')
|
self.file = tempfile.NamedTemporaryFile(delete=False, suffix='.txt')
|
||||||
self.file_path = pathlib.Path(self.file.name)
|
self.file_path = pathlib.Path(self.file.name)
|
||||||
self.message = Message(Question('This is a question'),
|
self.message = Message(Question('This is a question.'),
|
||||||
Answer('This is an answer'),
|
Answer('This is an answer.'),
|
||||||
{Tag('tag1'), Tag('tag2')},
|
{Tag('tag1'), Tag('tag2')},
|
||||||
self.file_path)
|
self.file_path)
|
||||||
|
|
||||||
@ -100,7 +100,12 @@ class MessageToFileTxtTestCase(CmmTestCase):
|
|||||||
|
|
||||||
with open(self.file_path, "r") as fd:
|
with open(self.file_path, "r") as fd:
|
||||||
content = fd.read()
|
content = fd.read()
|
||||||
expected_content = "TAGS: tag1 tag2\n=== QUESTION ===\nThis is a question\n=== ANSWER ===\nThis is an answer\n"
|
expected_content = """TAGS: tag1 tag2
|
||||||
|
=== QUESTION ===
|
||||||
|
This is a question.
|
||||||
|
=== ANSWER ===
|
||||||
|
This is an answer.
|
||||||
|
"""
|
||||||
self.assertEqual(content, expected_content)
|
self.assertEqual(content, expected_content)
|
||||||
|
|
||||||
def test_to_file_unsupported_file_type(self) -> None:
|
def test_to_file_unsupported_file_type(self) -> None:
|
||||||
@ -122,12 +127,55 @@ class MessageToFileTxtTestCase(CmmTestCase):
|
|||||||
self.message.file_path = self.file_path
|
self.message.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 = Message(Question('This is a question.'),
|
||||||
|
Answer('This is an answer.'),
|
||||||
|
{Tag('tag1'), Tag('tag2')},
|
||||||
|
self.file_path)
|
||||||
|
self.message_multiline = Message(Question('This is a\nmultiline question.'),
|
||||||
|
Answer('This is a\nmultiline answer.'),
|
||||||
|
{Tag('tag1'), Tag('tag2')},
|
||||||
|
self.file_path)
|
||||||
|
|
||||||
|
def tearDown(self) -> None:
|
||||||
|
self.file.close()
|
||||||
|
self.file_path.unlink()
|
||||||
|
|
||||||
|
def test_to_file_yaml(self) -> None:
|
||||||
|
self.message.to_file(self.file_path)
|
||||||
|
|
||||||
|
with open(self.file_path, "r") as fd:
|
||||||
|
content = fd.read()
|
||||||
|
expected_content = "answer: This is an answer.\nquestion: This is a question.\ntags:\n- tag1\n- tag2\n"
|
||||||
|
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 = """answer: |-
|
||||||
|
This is a
|
||||||
|
multiline answer.
|
||||||
|
question: |-
|
||||||
|
This is a
|
||||||
|
multiline question.
|
||||||
|
tags:
|
||||||
|
- tag1
|
||||||
|
- tag2
|
||||||
|
"""
|
||||||
|
self.assertEqual(content, expected_content)
|
||||||
|
|
||||||
|
|
||||||
class MessageFromFileTxtTestCase(CmmTestCase):
|
class MessageFromFileTxtTestCase(CmmTestCase):
|
||||||
def setUp(self) -> None:
|
def setUp(self) -> None:
|
||||||
self.file = tempfile.NamedTemporaryFile(delete=False, suffix='.txt')
|
self.file = tempfile.NamedTemporaryFile(delete=False, suffix='.txt')
|
||||||
self.file_path = pathlib.Path(self.file.name)
|
self.file_path = pathlib.Path(self.file.name)
|
||||||
with open(self.file_path, "w") as fd:
|
with open(self.file_path, "w") as fd:
|
||||||
fd.write("TAGS: tag1 tag2\n=== QUESTION ===\nThis is a question\n=== ANSWER ===\nThis is an answer\n")
|
fd.write("TAGS: tag1 tag2\n=== QUESTION ===\nThis is a question.\n=== ANSWER ===\nThis is an answer.\n")
|
||||||
|
|
||||||
def tearDown(self) -> None:
|
def tearDown(self) -> None:
|
||||||
self.file.close()
|
self.file.close()
|
||||||
@ -136,13 +184,39 @@ class MessageFromFileTxtTestCase(CmmTestCase):
|
|||||||
def test_from_file_txt(self) -> None:
|
def test_from_file_txt(self) -> None:
|
||||||
message = Message.from_file(self.file_path)
|
message = Message.from_file(self.file_path)
|
||||||
self.assertIsInstance(message, Message)
|
self.assertIsInstance(message, Message)
|
||||||
self.assertEqual(message.question, 'This is a question')
|
self.assertEqual(message.question, 'This is a question.')
|
||||||
self.assertEqual(message.answer, 'This is an answer')
|
self.assertEqual(message.answer, 'This is an answer.')
|
||||||
self.assertSetEqual(cast(set[Tag], message.tags), {Tag('tag1'), Tag('tag2')})
|
self.assertSetEqual(cast(set[Tag], message.tags), {Tag('tag1'), Tag('tag2')})
|
||||||
self.assertEqual(message.file_path, self.file_path)
|
self.assertEqual(message.file_path, self.file_path)
|
||||||
|
|
||||||
def test_from_file_not_exists(self) -> None:
|
def test_from_file_not_exists(self) -> None:
|
||||||
file_not_exists = pathlib.Path("example.doc")
|
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")
|
||||||
|
|
||||||
|
|
||||||
|
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("question: |-\n This is a question.\nanswer: |-\n This is an answer.\ntags:\n- tag1\n- tag2")
|
||||||
|
|
||||||
|
def tearDown(self) -> None:
|
||||||
|
self.file.close()
|
||||||
|
self.file_path.unlink()
|
||||||
|
|
||||||
|
def test_from_file_yaml(self) -> None:
|
||||||
|
message = Message.from_file(self.file_path)
|
||||||
|
self.assertIsInstance(message, Message)
|
||||||
|
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_not_exists(self) -> None:
|
||||||
|
file_not_exists = pathlib.Path("example.yaml")
|
||||||
with self.assertRaises(MessageError) as cm:
|
with self.assertRaises(MessageError) as cm:
|
||||||
Message.from_file(file_not_exists)
|
Message.from_file(file_not_exists)
|
||||||
self.assertEqual(str(cm.exception), f"Message file '{file_not_exists}' does not exist")
|
self.assertEqual(str(cm.exception), f"Message file '{file_not_exists}' does not exist")
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user