Merge pull request #10 from zenproducts/improve/remove-old-function

古い内部用クラスの削除

Author: hirokiky

setup.py

@@ -25,8 +25,11 @@
             "aiohttp",
         ],
         "dev": [
+            "aioresponses",
             "pytest",
+            "pytest-asyncio",
             "pytest-mock",
+            "pytest-responses",
             "mypy",
             "types-requests",
         ],

shodo/aio/lint.py

@@ -4,16 +4,16 @@
 import aiohttp
 
 from shodo.aio.api import lint_create, lint_result
-from shodo.lint import Lint, LintFailed, LintResult, Message
+from shodo.lint import LintFailed, LintResult, LintStatus, Message
 
 
-async def lint(body: str, is_html: bool = False, profile: Optional[str] = None) -> LintResult:
+async def lint(body: str, is_html: bool = False, profile: Optional[str] = None, _initial_pause: float = 0.25) -> LintResult:
     async with aiohttp.ClientSession() as session:
         create_res = await lint_create(body, is_html, profile, session)
 
-        status = Lint.STATUS_PROCESSING
-        pause = 0.25
-        while status == Lint.STATUS_PROCESSING:
+        status = LintStatus.PROCESSING.value
+        pause = _initial_pause
+        while status == LintStatus.PROCESSING.value:
             await asyncio.sleep(pause)
             result_res = await lint_result(create_res.lint_id, profile, session)
             status = result_res.status
@@ -23,7 +23,7 @@ async def lint(body: str, is_html: bool = False, profile: Optional[str] = None)
             if pause < 16:
                 pause *= 2
 
-        if status == Lint.STATUS_FAILED:
+        if status == LintStatus.FAILED.value:
             raise LintFailed
 
         return LintResult(status=status, messages=messages, updated=result_res.updated)

shodo/api.py

@@ -1,13 +1,20 @@
 import time
 from dataclasses import dataclass
-from datetime import datetime
+from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any, Dict, Generator, List, Optional
 
 import requests
 
 from shodo.conf import conf
 
+try:
+    from zoneinfo import ZoneInfo
+    JST = ZoneInfo("Asia/Tokyo")
+except ImportError:
+    from datetime import timedelta, timezone
+    JST = timezone(timedelta(hours=+9), "JST")
+
 
 def api_path(path, profile) -> str:
     return conf(profile).api_root.rstrip("/") + "/" + path.strip("/") + "/"
@@ -37,7 +44,7 @@ class LintResultResponse:
 
     def __post_init__(self) -> None:
         if isinstance(self.updated, int):
-            self.updated = datetime.fromtimestamp(self.updated)
+            self.updated = datetime.fromtimestamp(self.updated, JST)
 
 
 def lint_create(

shodo/lint.py

@@ -1,6 +1,7 @@
 import time
 from dataclasses import asdict, dataclass
 from datetime import datetime
+from enum import Enum
 from typing import Any, Dict, List, Optional
 
 from shodo.api import lint_create, lint_result
@@ -64,46 +65,18 @@ class LintFailed(Exception):
     pass
 
 
-class Lint:
-    STATUS_PROCESSING = "processing"
-    STATUS_FAILED = "failed"
+class LintStatus(Enum):
+    PROCESSING = "processing"
+    FAILED = "failed"
+    DONE = "done"
 
-    def __init__(self, body, lint_id, profile):
-        self.body = body
-        self.lint_id = lint_id
-        self.body = None
-        self.status = self.STATUS_PROCESSING
-        self.messages = []
-        self.profile = profile
 
-    def results(self):
-        while self.status == self.STATUS_PROCESSING:
-            time.sleep(0.5)
-            res = lint_result(self.lint_id, self.profile)
-            self.status = res.status
-            msgs = [Message.load(m) for m in res.messages]
-            self.messages = sorted(msgs, key=lambda m: (m.from_.line, m.from_.ch))
-
-        if self.status == self.STATUS_FAILED:
-            raise LintFailed
-
-        return self.messages
-
-    def __repr__(self):
-        return f"Lint({self.lint_id})"
-
-    @classmethod
-    def start(cls, body: str, is_html: bool = False, profile: Optional[str] = None):
-        res = lint_create(body, is_html, profile)
-        return cls(body, res.lint_id, profile)
-
-
-def lint(body: str, is_html: bool = False, profile: Optional[str] = None) -> LintResult:
+def lint(body: str, is_html: bool = False, profile: Optional[str] = None, _initial_pause: float=0.25) -> LintResult:
     create_res = lint_create(body, is_html, profile)
 
-    status = Lint.STATUS_PROCESSING
-    pause = 0.25
-    while status == Lint.STATUS_PROCESSING:
+    status = LintStatus.PROCESSING.value
+    pause = _initial_pause
+    while status == LintStatus.PROCESSING.value:
         time.sleep(pause)
         result_res = lint_result(create_res.lint_id, profile)
         status = result_res.status
@@ -113,7 +86,7 @@ def lint(body: str, is_html: bool = False, profile: Optional[str] = None) -> Lin
         if pause < 16:
             pause *= 2
 
-    if status == Lint.STATUS_FAILED:
+    if status == LintStatus.FAILED.value:
         raise LintFailed
 
     return LintResult(status=status, messages=messages, updated=result_res.updated)

shodo/main.py

@@ -10,7 +10,7 @@
 
 from shodo.api import download_image, list_post_files
 from shodo.conf import UnableLocateCredentialsError, save_credentials
-from shodo.lint import Lint
+from shodo.lint import lint as shodo_lint
 
 
 class ClickCatchExceptions(click.Group):
@@ -72,32 +72,28 @@ def lint(filename, html, output, profile):
     if not body:
         return
 
-    linting = Lint.start(body, is_html=html, profile=profile)
     click.echo("Linting...", err=True)
+    result = shodo_lint(body, is_html=html, profile=profile)
 
     if output == "json":
         click.echo(
             json.dumps(
-                [msg.asdict() for msg in linting.results()],
+                [msg.asdict() for msg in result.messages],
                 ensure_ascii=False,
                 indent=2,
             )
         )
         return
 
-    for message in linting.results():
+    for message in result.messages:
         if message.score < 0.5:
             continue
         color = "red" if message.severity == message.ERROR else "yellow"
         body_highlight = (
             body[message.index - 10 : message.index]
             + click.style(
                 body[message.index : message.index_to]
-                + (
-                    f"（→ {message.after or 'トル'}）"
-                    if message.after is not None
-                    else ""
-                ),
+                + (f"（→ {message.after or 'トル'}）" if message.after is not None else ""),
                 color,
             )
             + body[message.index_to : message.index_to + 10]
@@ -109,7 +105,7 @@ def lint(filename, html, output, profile):
         click.echo("     ", nl=False)
         click.echo(body_highlight)
 
-    if linting.messages:
+    if result.messages:
         sys.exit(1)
 
 

tests/conftest.py

@@ -0,0 +1,7 @@
+import pytest
+
+
+@pytest.fixture
+def credential(monkeypatch):
+    monkeypatch.setenv("SHODO_API_ROOT", "https://api.shodo.ink/@shodo/shodo/")
+    monkeypatch.setenv("SHODO_API_TOKEN", "test-token")

tests/test_aio/__init__.py

@@ -0,0 +1,43 @@
+from datetime import datetime, timezone
+from aioresponses import aioresponses
+
+import pytest
+from shodo.aio.lint import lint
+
+
+@pytest.mark.asyncio
+async def test_lint(credential):
+    with aioresponses() as m:
+        m.post(
+            "https://api.shodo.ink/@shodo/shodo/lint/",
+            payload={
+                "lint_id": "spam-spam-spam",
+                "monthly_amount": 100,
+                "current_usage": 10,
+                "len_body": 10,
+                "len_used": 10,
+            },
+        )
+        m.get(
+            "https://api.shodo.ink/@shodo/shodo/lint/spam-spam-spam/",
+            payload={"status": "done", "updated": 1_700_000_000, "messages": []},
+        )
+        actual = await lint("body", is_html=False)
+
+        assert actual.status == "done"
+        assert actual.messages == []
+        assert actual.updated.timetuple()[:6] == (2023, 11, 15, 7, 13, 20)
+        m.assert_called_with(
+            "https://api.shodo.ink/@shodo/shodo/lint/",
+            "post",
+            json={
+                "body": "body",
+                "type": "text",
+            },
+            headers={"Authorization": "Bearer test-token"},
+        )
+        m.assert_called_with(
+            "https://api.shodo.ink/@shodo/shodo/lint/spam-spam-spam/",
+            "get",
+            headers={"Authorization": "Bearer test-token"},
+        )

tests/test_aio/test_lint.py

@@ -0,0 +1,40 @@
+import json
+from datetime import datetime, timezone
+
+from shodo.lint import lint
+
+
+class TestLint:
+    def test_lint(self, credential, responses):
+        responses.add(
+            "POST",
+            "https://api.shodo.ink/@shodo/shodo/lint/",
+            json={
+                "lint_id": "spam-spam-spam",
+                "monthly_amount": 100,
+                "current_usage": 10,
+                "len_body": 10,
+                "len_used": 10,
+            },
+        )
+        responses.add(
+            "GET",
+            "https://api.shodo.ink/@shodo/shodo/lint/spam-spam-spam/",
+            json={
+                "status": "done",
+                "messages": [],
+                "updated": 1_700_000_000,
+            },
+        )
+
+        actual = lint("これはテストです", is_html=False, profile=None, _initial_pause=0)
+
+        assert actual.status == "done"
+        assert actual.messages == []
+        assert actual.updated.timetuple()[:6] == (2023, 11, 15, 7, 13, 20)
+
+        assert len(responses.calls) == 2
+        assert json.loads(responses.calls[0].request.body.decode("utf-8")) == {
+            "body": "これはテストです",
+            "type": "text",
+        }