test_app.py

import json
import os
import tempfile
import unittest
from pathlib import Path
from unittest.mock import Mock, patch

import app


RSS = """<?xml version="1.0"?>
<rss version="2.0"><channel><item>
<title>New AI agent model released</title>
<link>https://example.com/news/agent-model?utm_source=test</link>
<description>A new model for useful applications and benchmark research.</description>
<pubDate>Wed, 16 Sep 2026 09:00:00 GMT</pubDate>
</item></channel></rss>"""

INDEX = """<html><body><article><h2>
<a href="/news/agent-model">New AI agent model released</a>
</h2></article></body></html>"""

ARTICLE = """<html><head><title>New AI agent model released</title></head>
<body><main><p>This artificial intelligence model introduces a new agent architecture.</p>
<p>The research includes training, inference, open source weights, and benchmark results.</p></main></body></html>"""


class AppTests(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory()
        self.env = patch.dict(os.environ, {"WECHAT_AUTO_SHARE_DATA_DIR": self.temp.name})
        self.env.start()
        app.init_db()

    def tearDown(self):
        self.env.stop()
        self.temp.cleanup()

    def create_two_source_candidates(self):
        with app.connect() as conn:
            sources = {
                row["name"]: row
                for row in conn.execute(
                    "SELECT * FROM sources WHERE name IN ('OpenAI News', 'Google DeepMind')"
                )
            }
        app.store_candidate(
            sources["OpenAI News"],
            {"title": "OpenAI agent model released", "url": "https://example.com/openai-agent"},
            app.extract_article(ARTICLE)[1],
        )
        app.store_candidate(
            sources["Google DeepMind"],
            {"title": "DeepMind agent research released", "url": "https://example.com/deepmind-agent"},
            app.extract_article(ARTICLE)[1],
        )
        with app.connect() as conn:
            return [row[0] for row in conn.execute("SELECT id FROM candidates ORDER BY id")]

    def test_rss_parse_and_normalize(self):
        items = app.parse_rss(RSS, "https://example.com/feed.xml")
        self.assertEqual(len(items), 1)
        self.assertEqual(app.normalize_url(items[0]["url"]), "https://example.com/news/agent-model")

    def test_html_index_and_article_extract(self):
        items = app.parse_html_index(INDEX, "https://example.com/news")
        self.assertEqual(items[0]["url"], "https://example.com/news/agent-model")
        title, content = app.extract_article(ARTICLE)
        self.assertIn("agent model", title)
        self.assertIn("benchmark", content)

    def test_store_candidate_is_idempotent(self):
        with app.connect() as conn:
            source = conn.execute("SELECT * FROM sources WHERE kind = 'manual'").fetchone()
        item = {"title": "New AI agent model released", "url": "https://example.com/news/agent-model"}
        self.assertTrue(app.store_candidate(source, item, app.extract_article(ARTICLE)[1]))
        self.assertFalse(app.store_candidate(source, item, app.extract_article(ARTICLE)[1]))
        with app.connect() as conn:
            row = conn.execute(
                "SELECT general_copy, hardcore_copy FROM candidates"
            ).fetchone()
            count = conn.execute("SELECT COUNT(*) FROM candidates").fetchone()[0]
        self.assertEqual(count, 1)
        self.assertIn("AI", row["general_copy"])
        self.assertIn("技术", row["hardcore_copy"])

    def test_scan_falls_back_to_feed_summary(self):
        with app.connect() as conn:
            source = conn.execute("SELECT * FROM sources WHERE name = 'OpenAI News'").fetchone()
        with patch("app.fetch_text", side_effect=[RSS, ValueError("article blocked")]):
            result = app.scan_source(source, 1)
        self.assertEqual(result["added"], 1)
        self.assertIn("已使用摘要", result["error"])

    def test_local_page_renders_candidate(self):
        with app.connect() as conn:
            source = conn.execute("SELECT * FROM sources WHERE kind = 'manual'").fetchone()
        app.store_candidate(
            source,
            {"title": "New AI agent model released", "url": "https://example.com/news/agent-model"},
            app.extract_article(ARTICLE)[1],
        )
        client = app.create_web_app().test_client()
        response = client.get("/")
        self.assertEqual(response.status_code, 200)
        page = response.get_data(as_text=True)
        self.assertIn("New AI agent model released", page)
        self.assertIn("保存 AI 配置", page)
        self.assertIn("退出程序", page)

    def test_qr_route_returns_png(self):
        with app.connect() as conn:
            source = conn.execute("SELECT * FROM sources WHERE kind = 'manual'").fetchone()
        app.store_candidate(
            source,
            {"title": "New AI agent model released", "url": "https://example.com/news/agent-model"},
            app.extract_article(ARTICLE)[1],
        )
        with app.connect() as conn:
            candidate_id = conn.execute("SELECT id FROM candidates").fetchone()[0]
        response = app.create_web_app().test_client().get(f"/candidate/{candidate_id}/qr.png")
        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.mimetype, "image/png")
        self.assertTrue(response.data.startswith(b"\x89PNG\r\n\x1a\n"))

    def test_candidate_status_requires_csrf_and_hides_candidate(self):
        with app.connect() as conn:
            source = conn.execute("SELECT * FROM sources WHERE kind = 'manual'").fetchone()
        app.store_candidate(
            source,
            {"title": "New AI agent model released", "url": "https://example.com/news/agent-model"},
            app.extract_article(ARTICLE)[1],
        )
        with app.connect() as conn:
            candidate_id = conn.execute("SELECT id FROM candidates").fetchone()[0]
        client = app.create_web_app().test_client()
        self.assertEqual(
            client.post(f"/candidate/{candidate_id}/status", data={"status": "published"}).status_code,
            403,
        )
        response = client.post(
            f"/candidate/{candidate_id}/status",
            data={"status": "published", "csrf": app.get_setting("csrf_token")},
            follow_redirects=True,
        )
        self.assertEqual(response.status_code, 200)
        self.assertNotIn("New AI agent model released", response.get_data(as_text=True))
        with app.connect() as conn:
            status = conn.execute("SELECT status FROM candidates WHERE id = ?", (candidate_id,)).fetchone()[0]
        self.assertEqual(status, "published")

    def test_scheduled_action_runs_bounded_scan(self):
        action = app.scheduled_action()
        self.assertIn("scan --max-items 3", action)
        self.assertIn("app.py", action)

    def test_wecom_payload_requires_bounded_sender_scope(self):
        with app.connect() as conn:
            source = conn.execute("SELECT * FROM sources WHERE kind = 'manual'").fetchone()
        app.store_candidate(
            source,
            {"title": "New AI agent model released", "url": "https://example.com/news/agent-model"},
            app.extract_article(ARTICLE)[1],
        )
        with app.connect() as conn:
            candidate = conn.execute(
                "SELECT c.*, a.title, a.url FROM candidates c JOIN articles a ON a.id = c.article_id"
            ).fetchone()
        with self.assertRaisesRegex(ValueError, "至少填写"):
            app.build_wecom_moment_payload(candidate, "general", "", "", "")
        payload = app.build_wecom_moment_payload(
            candidate, "hardcore", "alice, bob", "2", "et-one"
        )
        self.assertEqual(payload["visible_range"]["sender_list"]["user_list"], ["alice", "bob"])
        self.assertEqual(payload["visible_range"]["sender_list"]["department_list"], [2])
        self.assertEqual(payload["visible_range"]["external_contact_list"]["tag_list"], ["et-one"])
        self.assertEqual(payload["attachments"][0]["link"]["url"], "https://example.com/news/agent-model")

    def test_wecom_submit_is_idempotent(self):
        with app.connect() as conn:
            source = conn.execute("SELECT * FROM sources WHERE kind = 'manual'").fetchone()
        app.store_candidate(
            source,
            {"title": "New AI agent model released", "url": "https://example.com/news/agent-model"},
            app.extract_article(ARTICLE)[1],
        )
        with app.connect() as conn:
            candidate_id = conn.execute("SELECT id FROM candidates").fetchone()[0]
        response = Mock()
        response.raise_for_status.return_value = None
        response.json.return_value = {"errcode": 0, "errmsg": "ok", "jobid": "job-123"}
        with patch("app.wecom_access_token", return_value="token"), patch(
            "app.httpx.post", return_value=response
        ) as post:
            first = app.submit_wecom_moment(candidate_id, "general", "alice", "", "")
            second = app.submit_wecom_moment(candidate_id, "general", "alice", "", "")
        self.assertFalse(first["duplicate"])
        self.assertTrue(second["duplicate"])
        self.assertEqual(second["jobid"], "job-123")
        self.assertEqual(post.call_count, 1)
        with app.connect() as conn:
            job = conn.execute("SELECT * FROM delivery_jobs").fetchone()
        self.assertEqual(job["status"], "submitted")
        self.assertEqual(job["external_id"], "job-123")
        self.assertNotIn("token", job["request_json"])

    def test_wecom_route_requires_explicit_confirmation(self):
        with app.connect() as conn:
            source = conn.execute("SELECT * FROM sources WHERE kind = 'manual'").fetchone()
        app.store_candidate(
            source,
            {"title": "New AI agent model released", "url": "https://example.com/news/agent-model"},
            app.extract_article(ARTICLE)[1],
        )
        with app.connect() as conn:
            candidate_id = conn.execute("SELECT id FROM candidates").fetchone()[0]
        response = app.create_web_app().test_client().post(
            f"/candidate/{candidate_id}/wecom",
            data={"csrf": app.get_setting("csrf_token"), "audience": "general", "user_list": "alice"},
        )
        self.assertEqual(response.status_code, 400)

    def test_official_draft_preview_edit_and_exports(self):
        candidate_ids = self.create_two_source_candidates()
        draft_id = app.create_official_draft(candidate_ids, "general")
        with app.connect() as conn:
            draft = conn.execute("SELECT * FROM drafts WHERE id = ?", (draft_id,)).fetchone()
        self.assertEqual(len(json.loads(draft["sources_json"])), 2)
        self.assertIn("参考来源", draft["html"])
        self.assertNotIn("<script>", app.markdown_body_html("<script>alert(1)</script>"))

        client = app.create_web_app().test_client()
        page = client.get(f"/draft/{draft_id}")
        self.assertEqual(page.status_code, 200)
        response = client.post(
            f"/draft/{draft_id}",
            data={
                "csrf": app.get_setting("csrf_token"),
                "title": "Edited topic",
                "summary": "Edited summary",
                "body": "## Findings\n\nA checked fact [1].",
            },
            follow_redirects=True,
        )
        self.assertEqual(response.status_code, 200)
        self.assertIn("Edited topic", response.get_data(as_text=True))
        for kind, mimetype in (("md", "text/markdown"), ("html", "text/html")):
            export = client.get(f"/draft/{draft_id}/export/{kind}")
            self.assertEqual(export.status_code, 200)
            self.assertEqual(export.mimetype, mimetype)
            self.assertIn(b"Edited topic", export.data)
        cover = client.get(f"/draft/{draft_id}/cover.svg")
        self.assertEqual(cover.status_code, 200)
        self.assertEqual(cover.mimetype, "image/svg+xml")

    def test_official_draft_requires_distinct_sources(self):
        with app.connect() as conn:
            source = conn.execute("SELECT * FROM sources WHERE name = 'OpenAI News'").fetchone()
        for index in (1, 2):
            app.store_candidate(
                source,
                {
                    "title": f"OpenAI model release {index}",
                    "url": f"https://example.com/openai-{index}",
                },
                app.extract_article(ARTICLE)[1],
            )
        with app.connect() as conn:
            candidate_ids = [row[0] for row in conn.execute("SELECT id FROM candidates ORDER BY id")]
        with self.assertRaisesRegex(ValueError, "两个不同来源"):
            app.create_official_draft(candidate_ids, "hardcore")

    def test_ai_official_draft_keeps_all_source_markers(self):
        self.create_two_source_candidates()
        with app.connect() as conn:
            rows = conn.execute(
                """
                SELECT a.title, a.url, a.content, s.name AS source_name
                FROM candidates c JOIN articles a ON a.id = c.article_id
                JOIN sources s ON s.id = a.source_id ORDER BY c.id
                """
            ).fetchall()
        response = Mock()
        response.raise_for_status.return_value = None
        response.json.return_value = {
            "choices": [
                {
                    "message": {
                        "content": json.dumps(
                            {
                                "title": "Two-source AI update",
                                "summary": "A checked summary",
                                "introduction": "First report [1], second report [2].",
                                "sections": [{"heading": "What changed", "body": "Compared evidence [1][2]."}],
                                "conclusion": "More verification is still needed [1][2].",
                            }
                        )
                    }
                }
            ]
        }
        with patch("app.ai_config", return_value=("https://api.example/v1", "model", "key")), patch(
            "app.httpx.post", return_value=response
        ):
            result = app.generate_draft_content(rows, "hardcore")
        self.assertEqual(result["title"], "Two-source AI update")
        self.assertIn("[1]", result["body"])
        self.assertIn("[2]", result["body"])

    def test_web_ai_config_uses_keyring(self):
        client = app.create_web_app().test_client()
        with patch("app.validate_public_url", return_value="https://api.example/v1"), patch(
            "app.keyring.set_password"
        ) as save_secret:
            response = client.post(
                "/ai/config",
                data={
                    "csrf": app.get_setting("csrf_token"),
                    "base_url": "https://api.example/v1",
                    "model": "model-name",
                    "api_key": "secret-key",
                },
            )
        self.assertEqual(response.status_code, 302)
        self.assertEqual(app.get_setting("ai_model"), "model-name")
        save_secret.assert_called_once_with(app.KEYRING_SERVICE, "ai-api-key", "secret-key")

    def test_schedule_changes_only_after_valid_post(self):
        client = app.create_web_app().test_client()
        self.assertEqual(client.post("/schedule", data={"action": "install"}).status_code, 403)
        with patch("app.install_schedule") as install:
            response = client.post(
                "/schedule",
                data={"csrf": app.get_setting("csrf_token"), "action": "install"},
            )
        self.assertEqual(response.status_code, 302)
        install.assert_called_once_with()


if __name__ == "__main__":
    unittest.main()