refactor: migrate visitor/main to typed models with live API contract tests

Author: chazeon

tests/4/test_id3_tagging_with_missing_id3.py

@@ -8,6 +8,7 @@
 sys.path.insert(0, str(TESTS_DIR.parent))
 sys.path.insert(0, str(TESTS_DIR.parent / "vistopia"))
 from vistopia.visitor import Visitor
+from vistopia.models import RetagArticle, RetagSeries, Catalog
 
 CURR_TEST_DIR = Path(__file__).parent
 
@@ -36,21 +37,25 @@ def test_id3_tagging_with_missing_id3(tmpdir):
 
     Visitor.retag(
         test_mp3,
-        article_info={
-            "title": "测试标题",
-            "sort_number": "7",
-            "content_url": "http://example.com",
-        },
-        series_info={
-            "title": "测试系列",
-            "author": "测试作者",
-        },
-        catalog_info={}
+        article_info=RetagArticle(
+            title="测试标题",
+            sort_number="7",
+            content_url="http://example.com",
+        ),
+        series_info=RetagSeries(
+            title="测试系列",
+            author="测试作者",
+        ),
+        catalog_info=Catalog(
+            author="测试作者",
+            title="测试系列",
+            type="charge",
+            catalog=[],
+        )
     )
 
     tag = EasyID3(test_mp3)
 
     assert tag["title"] == ["测试标题"]
     assert tag["album"] == ["测试系列"]
     assert tag["artist"] == ["测试作者"]
-

tests/test_live_endpoints.py

@@ -0,0 +1,53 @@
+"""Live API contract checks for public (no-token) endpoints.
+
+These tests intentionally hit the real Vistopia API to verify that the
+observed response shapes still match our pydantic models.
+"""
+
+from urllib.parse import urljoin
+
+import requests
+
+from vistopia.models import Catalog, ContentShow, SearchResult, SubscriptionsList, validate_model
+
+
+BASE_URL = "https://api.vistopia.com.cn/api/v1/"
+TIMEOUT = 20
+
+
+def _get_data(uri: str, **params):
+    query = {"api_token": ""}
+    query.update(params)
+    response = requests.get(urljoin(BASE_URL, uri), params=query, timeout=TIMEOUT)
+    payload = response.json()
+    assert payload["status"] == "success"
+    assert "data" in payload
+    return payload["data"]
+
+
+def test_live_catalog_endpoint_matches_model():
+    data = _get_data("content/catalog/18")
+    model = validate_model(Catalog, data)
+    assert model.title
+    assert model.catalog
+    assert model.catalog[0].part
+
+
+def test_live_content_show_endpoint_matches_model():
+    data = _get_data("content/content-show/18")
+    model = validate_model(ContentShow, data)
+    assert model.title
+    assert model.author
+
+
+def test_live_search_endpoint_matches_model():
+    data = _get_data("search/web", keyword="八分")
+    model = validate_model(SearchResult, data)
+    assert isinstance(model.data, list)
+    assert len(model.data) > 0
+
+
+def test_live_subscriptions_endpoint_matches_model():
+    data = _get_data("user/subscriptions-list")
+    model = validate_model(SubscriptionsList, data)
+    assert isinstance(model.data, list)

tests/test_models.py

@@ -6,10 +6,8 @@
     Catalog,
     CatalogPart,
     ContentShow,
-    SearchPage,
     SearchItem,
     SearchResult,
-    SubscriptionsPage,
     SubscriptionItem,
     SubscriptionsList,
     validate_model,
@@ -30,9 +28,8 @@ def test_models_cover_all_keys_used_by_runtime_code():
     assert {"author", "title"} <= _field_names(ContentShow)
     assert {"data"} <= _field_names(SubscriptionsList)
     assert {"content_id", "title", "subtitle"} <= _field_names(SubscriptionItem)
-    assert {"data"} <= _field_names(SearchResult)
-    assert {"data", "current_page", "from_", "last_page"} <= _field_names(SearchPage)
-    assert {"data", "current_page", "from_", "last_page"} <= _field_names(SubscriptionsPage)
+    assert {"data", "current_page", "from_", "last_page"} <= _field_names(SearchResult)
+    assert {"data", "current_page", "from_", "last_page"} <= _field_names(SubscriptionsList)
     assert {"id", "author", "title", "share_desc", "data_type", "subtitle"} <= _field_names(SearchItem)
 
 
@@ -67,31 +64,29 @@ def test_catalog_model_parses_nested_structure():
 
 def test_search_result_coerces_numeric_id_from_string():
     payload = {
-        "data": {
-            "current_page": 1,
-            "from": 1,
-            "last_page": 1,
-            "next_page_url": None,
-            "per_page": 20,
-            "prev_page_url": None,
-            "to": 1,
-            "total": 1,
-            "data": [
-                {
-                    "id": "11",
-                    "author": "梁文道",
-                    "title": "八分",
-                    "share_desc": "知识只求八分饱",
-                    "data_type": "content",
-                    "subtitle": "",
-                }
-            ],
-        }
+        "current_page": 1,
+        "from": 1,
+        "last_page": 1,
+        "next_page_url": None,
+        "per_page": 20,
+        "prev_page_url": None,
+        "to": 1,
+        "total": 1,
+        "data": [
+            {
+                "id": "11",
+                "author": "梁文道",
+                "title": "八分",
+                "share_desc": "知识只求八分饱",
+                "data_type": "content",
+                "subtitle": "",
+            }
+        ],
     }
 
     model = validate_model(SearchResult, payload)
-    assert model.data.data[0].id == 11
-    assert model.data.from_ == 1
+    assert model.data[0].id == 11
+    assert model.from_ == 1
 
 
 def test_catalog_model_raises_on_missing_required_fields():

tests/test_visitor.py

@@ -16,16 +16,16 @@ def visitor():
 def test_get_catalog(visitor):
     catalog = visitor.get_catalog(id=11)
     assert catalog
-    assert catalog.get("author") == "梁文道"
-    assert catalog.get("title") == "八分"
+    assert catalog.author == "梁文道"
+    assert catalog.title == "八分"
 
-    assert isinstance(catalog.get("catalog"), list)
+    assert isinstance(catalog.catalog, list)
 
 
 def test_get_content_show(visitor):
     content_show = visitor.get_content_show(id=11)
-    assert content_show.get("author") == "梁文道"
-    assert content_show.get("title") == "八分"
+    assert content_show.author == "梁文道"
+    assert content_show.title == "八分"
 
 
 def test_save_show(visitor, tmpdir):
@@ -75,13 +75,13 @@ def test_search(visitor: Visitor, keyword: str, expected: tuple):
     data = visitor.search(keyword)
     assert isinstance(data, list)
     for k in ("id", "author", "title", "share_desc", "data_type"):
-        assert all([k in item.keys() for item in data])
+        assert all([hasattr(item, k) for item in data])
     assert any([
         (
-            int(item["id"]),
-            item["author"],
-            item["title"],
-            item["share_desc"],
-            item["data_type"],
+            int(item.id),
+            item.author,
+            item.title,
+            item.share_desc,
+            item.data_type,
         ) == expected for item in data
     ])

vistopia/main.py

@@ -10,6 +10,7 @@
 from .visitor import Visitor
 from .utils import range_expand
 from .__version__ import __version__
+from .models import dump_model
 
 logger = getLogger(__name__)
 
@@ -51,19 +52,19 @@ def main(ctx: click.Context, **argv):
 def search(ctx: click.Context, **argv):
     visitor: Visitor = ctx.obj.visitor
     search_result_list = visitor.search(argv.pop("keyword"))
-    logger.debug(json.dumps(search_result_list, indent=2, ensure_ascii=False))
+    logger.debug(json.dumps([dump_model(item) for item in search_result_list], indent=2, ensure_ascii=False))
 
     table = []
     for item in search_result_list:
-        if item["data_type"] != "content":
+        if item.data_type != "content":
             continue
-        author = item["author"]
-        if item["subtitle"]:
-            title = "%s: %s" % ([item["title"], item["subtitle"]])
+        author = item.author
+        if item.subtitle:
+            title = "%s: %s" % ([item.title, item.subtitle])
         else:
-            title = item["title"]
-        desc = item["share_desc"]
-        content_id = item["id"]
+            title = item.title
+        desc = item.share_desc
+        content_id = item.id
         table.append((content_id, author, title, desc))
 
     click.echo(tabulate(table))
@@ -78,8 +79,8 @@ def subscriptions(ctx: click.Context):
 
     table = []
     for show in visitor.get_user_subscriptions_list():
-        title = ": ".join([show["title"], show["subtitle"]])
-        content_id = show["content_id"]
+        title = ": ".join([show.title, show.subtitle])
+        content_id = show.content_id
         table.append((content_id, title))
 
     click.echo(tabulate(table))
@@ -94,27 +95,27 @@ def show_content(ctx: click.Context, **argv):
     content_id = argv.pop("id")
     logger.debug(visitor.get_content_show(content_id))
     logger.debug(
-        json.dumps(visitor.get_catalog(content_id), indent=2, ensure_ascii=False)
+        json.dumps(dump_model(visitor.get_catalog(content_id)), indent=2, ensure_ascii=False)
     )
 
     catalog = visitor.get_catalog(content_id)
 
-    click.echo(f"{catalog['title']}")
+    click.echo(f"{catalog.title}")
     click.echo()
-    click.echo(f"艺人: {catalog['author']}")
-    click.echo(f"类型: {catalog['type']}")
+    click.echo(f"艺人: {catalog.author}")
+    click.echo(f"类型: {catalog.type}")
     click.echo()
 
-    for part in catalog["catalog"]:
-        click.echo(f"{part['catalog_number']}  {part['catalog_title']}")
+    for part in catalog.catalog:
+        click.echo(f"{part.catalog_number}  {part.catalog_title}")
         table = []
-        for article in part["part"]:
+        for article in part.part:
             table.append(
                 (
-                    article["sort_number"],
+                    article.sort_number,
                     # article["article_id"],
-                    article["title"],
-                    article["duration_str"],
+                    article.title,
+                    article.duration_str,
                 )
             )
         click.echo(tabulate(table))
@@ -132,7 +133,7 @@ def save_show(ctx: click.Context, **argv):
 
     logger.debug(
         json.dumps(
-            ctx.obj.visitor.get_catalog(content_id), indent=2, ensure_ascii=False
+            dump_model(ctx.obj.visitor.get_catalog(content_id)), indent=2, ensure_ascii=False
         )
     )
 
@@ -166,7 +167,7 @@ def save_transcript(ctx: click.Context, **argv):
 
     logger.debug(
         json.dumps(
-            ctx.obj.visitor.get_catalog(content_id), indent=2, ensure_ascii=False
+            dump_model(ctx.obj.visitor.get_catalog(content_id)), indent=2, ensure_ascii=False
         )
     )