copykar/tests/test_websites_and_dispatch_order.py

import asyncio
import time
import json
from unittest.mock import AsyncMock, MagicMock, patch
from db.database import init_db
from db.repository import Repository
from db.models import TargetChannel, SourceWebsite
from core.queue import RedisQueue
from services.website_analyzer import WebsiteAnalyzer
from services.website_collector import WebsiteCollectorService, _clean_html, _extract_nested


async def test_target_dispatch_order_and_queue():
    await init_db()
    repo = Repository()

    # 1. Target creation and dispatch_order field
    unique_channel_id = -10088776655 - int(time.time() % 100000)
    t_id = await repo.add_target(channel_id=unique_channel_id, title="Dispatch Test Target", username="disp_test")
    assert t_id is not None

    target = await repo.get_target_by_id(t_id)
    assert target.dispatch_order == "order"

    # Toggle to random
    await repo.update_target_dispatch_order(t_id, "random")
    target_updated = await repo.get_target_by_id(t_id)
    assert target_updated.dispatch_order == "random"

    # 2. Redis queue FIFO vs Random pop test
    queue = RedisQueue()
    await queue.connect()

    # Clear queue for test target
    key = queue._get_target_key(t_id)
    await queue.client.delete(key)

    # Push 3 items: Item 1, Item 2, Item 3
    await queue.push_target_post(t_id, {"post_id": 1, "text": "Post 1"})
    await queue.push_target_post(t_id, {"post_id": 2, "text": "Post 2"})
    await queue.push_target_post(t_id, {"post_id": 3, "text": "Post 3"})

    assert await queue.get_target_queue_size(t_id) == 3

    # Random pop should retrieve one of the 3 items and leave 2
    popped_random = await queue.pop_target_post(t_id, dispatch_order="random")
    assert popped_random is not None
    assert popped_random["post_id"] in (1, 2, 3)
    assert await queue.get_target_queue_size(t_id) == 2

    # Clean up
    await queue.client.delete(key)
    await repo.delete_target(t_id)
    await queue.close()


async def test_source_websites_crud_and_collector():
    await init_db()
    repo = Repository()

    # 1. Add Source Website
    test_url = f"https://example.com/blog-{int(time.time())}"
    site_id = await repo.add_source_website(
        name="Example Tech Blog",
        url=test_url,
        check_interval_min=15,
        auto_reanalyze_hours=12,
        api_config={
            "endpoint_url": f"{test_url}/wp-json/wp/v2/posts",
            "parser_type": "wordpress_json",
            "field_mappings": {
                "title": "title.rendered",
                "content": "content.rendered",
                "link": "link"
            }
        }
    )
    assert site_id is not None

    site = await repo.get_source_website_by_id(site_id)
    assert site.name == "Example Tech Blog"
    assert site.check_interval_min == 15
    assert site.auto_reanalyze_hours == 12
    assert site.api_config.get("parser_type") == "wordpress_json"

    # 2. Update intervals and status
    await repo.update_source_website_interval(site_id, 45)
    await repo.update_source_website_reanalyze_hours(site_id, 48)
    await repo.update_source_website_fetch_status(site_id, error="Connection timeout test")

    site_err = await repo.get_source_website_by_id(site_id)
    assert site_err.check_interval_min == 45
    assert site_err.auto_reanalyze_hours == 48
    assert site_err.last_error == "Connection timeout test"

    # 3. Clean up error
    await repo.update_source_website_fetch_status(site_id, error=None)
    site_ok = await repo.get_source_website_by_id(site_id)
    assert site_ok.last_error is None
    assert site_ok.last_fetched_at is not None

    # Clean up
    await repo.delete_source_website(site_id)


async def test_website_analyzer_and_collector_mock():
    # Helper functions test
    assert _clean_html("<p>Hello <b>World</b>!</p>") == "Hello World!"
    sample_dict = {"channel": {"items": [{"title": "News 1"}]}}
    assert _extract_nested(sample_dict, "channel.items.0.title") == "News 1"

    # Website Analyzer mock test
    llm_mock = MagicMock()
    llm_mock.generate_json = AsyncMock(return_value={
        "status": "success",
        "endpoint_url": "https://digiato.com/wp-json/wp/v2/posts?per_page=10",
        "parser_type": "wordpress_json",
        "field_mappings": {
            "title": "title.rendered",
            "content": "content.rendered",
            "link": "link"
        },
        "summary": "سایت از وردپرس استفاده می‌کند."
    })

    analyzer = WebsiteAnalyzer(llm=llm_mock)
    with patch("httpx.AsyncClient.get") as mock_get:
        mock_resp = MagicMock()
        mock_resp.status_code = 200
        mock_resp.text = '<html><head><link rel="https://api.w.org/" href="https://digiato.com/wp-json/" /></head></html>'
        mock_get.return_value = mock_resp

        ok, cfg, summary = await analyzer.analyze_website("https://digiato.com")
        assert ok is True
        assert cfg["parser_type"] == "wordpress_json"
        assert cfg["endpoint_url"] == "https://digiato.com/wp-json/wp/v2/posts?per_page=10"


async def main():
    await test_target_dispatch_order_and_queue()
    await test_source_websites_crud_and_collector()
    await test_website_analyzer_and_collector_mock()
    print("All Source Websites, Dispatch Order, and Markdown styling tests passed successfully!")


if __name__ == "__main__":
    asyncio.run(main())