| import json | |
| import asyncio | |
| from src.utils import ( | |
| format_registration_days, | |
| get_link_unique_key, | |
| safe_get, | |
| save_to_jsonl, | |
| ) | |
| def test_safe_get_nested_and_default(): | |
| data = {"a": {"b": [{"c": "value"}]}} | |
| assert asyncio.run(safe_get(data, "a", "b", 0, "c")) == "value" | |
| assert asyncio.run(safe_get(data, "a", "b", 1, "c", default="missing")) == "missing" | |
| def test_format_registration_days(): | |
| assert format_registration_days(400).startswith("\u6765\u95f2\u9c7c") | |
| assert format_registration_days(-1) == "\u672a\u77e5" | |
| def test_get_link_unique_key(): | |
| link = "https://www.goofish.com/item?id=123&foo=bar" | |
| assert get_link_unique_key(link) == "https://www.goofish.com/item?id=123" | |
| def test_save_to_jsonl(tmp_path, monkeypatch): | |
| monkeypatch.chdir(tmp_path) | |
| record = {"id": 1, "title": "Sony A7M4"} | |
| ok = asyncio.run(save_to_jsonl(record, keyword="sony a7m4")) | |
| assert ok is True | |
| output = tmp_path / "jsonl" / "sony_a7m4_full_data.jsonl" | |
| assert output.exists() | |
| lines = output.read_text(encoding="utf-8").splitlines() | |
| assert json.loads(lines[0]) == record | |