{
"event": "PreToolUse",
"tool_name": "Write",
"tool_input": {
"file_path": "\/var\/www\/dev.campus.systemische-tools.de\/crawler\/tests\/integration\/infrastructure\/test_db_url_queue.py",
"content": "from __future__ import annotations\n\nfrom datetime import datetime\n\nfrom src.domain.entities.crawl import Crawl\nfrom src.domain.value_objects.enums import CrawlMode, CrawlStatus, TriggerSource\nfrom src.domain.value_objects.normalized_url import normalize\nfrom src.infrastructure.persistence.connection import MariaDbConnectionFactory\nfrom src.infrastructure.persistence.crawl_repository import CrawlRepository\nfrom src.infrastructure.queue.db_url_queue import DbUrlQueue\n\n\ndef _seed_crawl(factory: MariaDbConnectionFactory) -> int:\n saved = CrawlRepository(factory).save(\n Crawl(\n id=None,\n base_url=\"https:\/\/campus-am-see.de\/\",\n mode=CrawlMode.FAST,\n started_at=datetime(2026, 4, 21),\n finished_at=None,\n status=CrawlStatus.RUNNING,\n trigger_source=TriggerSource.CLI,\n config={},\n ),\n )\n assert saved.id is not None\n return saved.id\n\n\ndef test_enqueue_dedupes_and_dequeue_returns_fifo(\n connection_factory: MariaDbConnectionFactory,\n) -> None:\n crawl_id = _seed_crawl(connection_factory)\n queue = DbUrlQueue(connection_factory)\n url_a = normalize(\"https:\/\/campus-am-see.de\/a\")\n url_b = normalize(\"https:\/\/campus-am-see.de\/b\")\n\n assert queue.enqueue(crawl_id, url_a)\n assert queue.enqueue(crawl_id, url_b)\n assert not queue.enqueue(crawl_id, url_a) # dedupe\n assert queue.pending_count(crawl_id) == 2\n\n claimed = queue.dequeue(crawl_id)\n assert claimed is not None\n assert claimed.url_hash == url_a.url_hash\n queue.mark_done(crawl_id, claimed)\n\n assert queue.pending_count(crawl_id) == 1\n"
}
}
{
"tool_response": {
"type": "create",
"filePath": "\/var\/www\/dev.campus.systemische-tools.de\/crawler\/tests\/integration\/infrastructure\/test_db_url_queue.py",
"content": "from __future__ import annotations\n\nfrom datetime import datetime\n\nfrom src.domain.entities.crawl import Crawl\nfrom src.domain.value_objects.enums import CrawlMode, CrawlStatus, TriggerSource\nfrom src.domain.value_objects.normalized_url import normalize\nfrom src.infrastructure.persistence.connection import MariaDbConnectionFactory\nfrom src.infrastructure.persistence.crawl_repository import CrawlRepository\nfrom src.infrastructure.queue.db_url_queue import DbUrlQueue\n\n\ndef _seed_crawl(factory: MariaDbConnectionFactory) -> int:\n saved = CrawlRepository(factory).save(\n Crawl(\n id=None,\n base_url=\"https:\/\/campus-am-see.de\/\",\n mode=CrawlMode.FAST,\n started_at=datetime(2026, 4, 21),\n finished_at=None,\n status=CrawlStatus.RUNNING,\n trigger_source=TriggerSource.CLI,\n config={},\n ),\n )\n assert saved.id is not None\n return saved.id\n\n\ndef test_enqueue_dedupes_and_dequeue_returns_fifo(\n connection_factory: MariaDbConnectionFactory,\n) -> None:\n crawl_id = _seed_crawl(connection_factory)\n queue = DbUrlQueue(connection_factory)\n url_a = normalize(\"https:\/\/campus-am-see.de\/a\")\n url_b = normalize(\"https:\/\/campus-am-see.de\/b\")\n\n assert queue.enqueue(crawl_id, url_a)\n assert queue.enqueue(crawl_id, url_b)\n assert not queue.enqueue(crawl_id, url_a) # dedupe\n assert queue.pending_count(crawl_id) == 2\n\n claimed = queue.dequeue(crawl_id)\n assert claimed is not None\n assert claimed.url_hash == url_a.url_hash\n queue.mark_done(crawl_id, claimed)\n\n assert queue.pending_count(crawl_id) == 1\n",
"structuredPatch": [],
"originalFile": null,
"userModified": false
}
}