update

Author: hsz1273327

.github/workflows/python-package.yml

@@ -0,0 +1,41 @@
+# This workflow will install Python dependencies, run tests and lint with a variety of Python versions
+# For more information see: https://help.github.com/actions/language-and-framework-guides/using-python-with-github-actions
+
+name: Python package
+
+on:
+  push:
+    branches: [master,dev]
+  pull_request:
+    branches: [master]
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        python-version: [3.6, 3.7, 3.8, 3.9]
+
+    steps:
+      - uses: actions/checkout@v2
+      - name: Set up Python ${{ matrix.python-version }}
+        uses: actions/setup-python@v2
+        with:
+          python-version: ${{ matrix.python-version }}
+      - name: Install devDependence
+        run: |
+          python -m pip install --upgrade pip
+          pip install mypy pycodestyle coverage lxml
+      - name: Install dependencies
+        run: |
+          if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
+      - name: Lint with pep8
+        run: |
+          pycodestyle --max-line-length=140 --ignore=E501 --first --statistics sserender
+      - name: Type Hint Check
+        run: |
+          mypy --ignore-missing-imports --show-column-numbers --follow-imports=silent --check-untyped-defs --disallow-untyped-defs --no-implicit-optional --warn-unused-ignores sserender
+      - name: Unit Test
+        run: |
+          python -m coverage run --source=sserender -m unittest discover -v -s . -p *test*.py
+          python -m coverage report

.gitignore

@@ -127,3 +127,4 @@ dmypy.json
 
 # Pyre type checker
 .pyre/
+doc_unittest

pmfprc.json

@@ -1,3 +1,6 @@
 {
-    "env": "venv"
+    "env": "venv",
+    "added_components": {
+        "cp_python_test//simplemodule_tests": "tests"
+    }
 }

sserender/__init__.py

@@ -1 +1,2 @@
 from .sse import SSE
+__all__ = ["SSE"]

sserender/sse.py

@@ -29,16 +29,41 @@ def __init__(self, ID: Optional[str] = None,
                  data: Optional[str] = None,
                  comment: Optional[str] = None,
                  retry: Optional[int] = None) -> None:
-        if any([ID, event, data, comment, retry]):
+        """构造sse消息.
+
+        除了retry外至少要有1个参数.
+
+        Args:
+            ID (Optional[str], optional): 消息id. Defaults to None.
+            event (Optional[str], optional): 消息事件. Defaults to None.
+            data (Optional[str], optional): 消息数据. Defaults to None.
+            comment (Optional[str], optional): 消息注释. Defaults to None.
+            retry (Optional[int], optional): 重连间隔时间,单位ms. Defaults to None.
+
+        Raises:
+            TypeError: [description]
+            ValueError: [description]
+        """
+        if any([ID, event, data, comment]):
             self.ID = ID
             self.event = event
             self.data = data
             self.retry = retry
             self.comment = comment
+            if self.retry and not isinstance(self.retry, int):
+                raise TypeError("retry argument must be int")
         else:
-            raise ValueError("at least one args")
+            raise ValueError("at least one argument")
+
+    def render(self, *, with_encode: bool = False) -> Union[str, bytes]:
+        """渲染sse消息对象
 
-    def render(self, with_encode: bool = False) -> Union[str, bytes]:
+        Args:
+            with_encode (bool, optional): 是否编码. Defaults to False.
+
+        Returns:
+            Union[str, bytes]: with_encode为True则返回为bytes否则为str
+        """
         buffer = io.StringIO()
         if self.comment is not None:
             for chunk in self._LINE_SEP_EXPR.split(self.comment):
@@ -57,8 +82,6 @@ def render(self, with_encode: bool = False) -> Union[str, bytes]:
                 buffer.write(self._DEFAULT_SEPARATOR)
 
         if self.retry is not None:
-            if not isinstance(self.retry, int):
-                raise TypeError("retry argument must be int")
             buffer.write("retry: {}".format(self.retry))
             buffer.write(self._DEFAULT_SEPARATOR)
 
@@ -69,7 +92,19 @@ def render(self, with_encode: bool = False) -> Union[str, bytes]:
         return result
 
     @classmethod
-    def from_content(clz, content: Union[str, bytes], strict: bool = False) -> "SSE":
+    def from_content(clz, content: Union[str, bytes], *, strict: bool = False) -> "SSE":
+        """从字符串或者字节流中解析出sse消息.
+
+        Args:
+            content (Union[str, bytes]): 待解析的内容
+            strict (bool, optional): 严格模式,校验结尾是否为`\r\n\r\n`. Defaults to False.
+
+        Raises:
+            ValueError: not end with `\r\n\r\n`
+
+        Returns:
+            [SSE]: sse消息对象
+        """
         if isinstance(content, bytes):
             content = content.decode("utf-8")
         if strict:
@@ -82,6 +117,7 @@ def from_content(clz, content: Union[str, bytes], strict: bool = False) -> "SSE"
         retry = 0
         comment = ""
         lines = content.strip().split(clz._DEFAULT_SEPARATOR)
+        print(lines)
         for line in lines:
             if line.startswith(":"):
                 c = line[1:].strip() + "\n"

tests/__init__.py

@@ -0,0 +1,6 @@
+def setUpModule() -> None:
+    print("[Module sserender Test Start]")
+
+
+def tearDownModule() -> None:
+    print("[Module sserender Test End]")

tests/test_module.py

@@ -0,0 +1,59 @@
+import unittest
+from pathlib import Path
+
+from sserender import SSE
+
+
+class ExampleTest(unittest.TestCase):
+
+    def setUp(self) -> None:
+        print("section start")
+
+    def tearDown(self) -> None:
+        print("section end")
+
+    def test_sse_render(self) -> None:
+        instance = SSE(ID="1", event="test", data="test_data", comment="test_comment", retry=12345)
+        res = instance.render()
+        assert ": test_comment" in res
+        assert "id: 1" in res
+        assert "event: test" in res
+        assert "data: test_data" in res
+        assert "retry: 12345" in res
+
+    def test_sse_render_to_bytes(self) -> None:
+        instance = SSE(ID="1", event="test", data="test_data", comment="test_comment", retry=12345)
+        res = instance.render(with_encode=True).decode("utf-8")
+        assert ": test_comment" in res
+        assert "id: 1" in res
+        assert "event: test" in res
+        assert "data: test_data" in res
+        assert "retry: 12345" in res
+
+    def test_sse_render_empty(self) -> None:
+        with self.assertRaisesRegex(ValueError, r"at least one argument") as a:
+            SSE()
+
+    def test_sse_render_retry_not_int(self) -> None:
+        with self.assertRaisesRegex(TypeError, r"retry argument must be int") as a:
+            SSE(ID="12", retry="12345")
+
+    def test_sse_from_content(self) -> None:
+        instance = SSE(ID="1", event="test", data="test_data", comment="test_comment", retry=12345)
+        content = instance.render()
+        sse = SSE.from_content(content)
+        print(sse.ID)
+        assert sse.ID == "1"
+
+    def test_sse_from_content_bytes(self) -> None:
+        instance = SSE(ID="1", event="test", data="test_data", comment="test_comment", retry=12345)
+        content = instance.render(with_encode=True)
+        sse = SSE.from_content(content)
+        print(sse.ID)
+        assert sse.ID == "1"
+
+    def test_sse_from_content_strict(self) -> None:
+        instance = SSE(ID="1", event="test", data="test_data", comment="test_comment", retry=12345)
+        content = instance.render().strip()
+        with self.assertRaisesRegex(ValueError, r"not end with") as a:
+            SSE.from_content(content, strict=True)