{
"event": "PreToolUse",
"tool_name": "Write",
"tool_input": {
"file_path": "\/var\/www\/dev.campus.systemische-tools.de\/crawler\/tests\/unit\/application\/test_check_external_link.py",
"content": "from __future__ import annotations\n\nfrom collections.abc import Mapping\nfrom datetime import datetime, timedelta\n\nimport pytest\n\nfrom src.application.use_cases.check_external_link import CheckExternalLinkUseCase\nfrom src.domain.ports.http_fetcher_port import FetchResult, HttpFetcherPort\nfrom src.domain.services.url_normalizer import UrlNormalizer\nfrom src.domain.value_objects.normalized_url import NormalizedUrl\nfrom tests.unit.application.fakes import FakeClock, FakeExternalCheckRepo\n\n\nclass _StubFetcher(HttpFetcherPort):\n def __init__(self, status: int, error: str | None = None) -> None:\n self.calls = 0\n self._status = status\n self._error = error\n\n async def fetch(self, url: NormalizedUrl) -> FetchResult:\n self.calls += 1\n headers: Mapping[str, str] = {}\n return FetchResult(\n final_url=url.url,\n status_code=self._status,\n redirect_chain=(),\n headers=headers,\n body=b\"\",\n response_time_ms=12,\n ttfb_ms=None,\n error=self._error,\n )\n\n\n@pytest.mark.asyncio\nasync def test_first_check_fetches_and_persists() -> None:\n fetcher = _StubFetcher(status=200)\n repo = FakeExternalCheckRepo()\n now = datetime(2026, 4, 21, 12, 0, 0)\n uc = CheckExternalLinkUseCase(\n fetcher=fetcher, checks=repo, normalizer=UrlNormalizer(),\n clock=FakeClock(now), cache_ttl=timedelta(hours=1),\n )\n\n result = await uc.execute(\"https:\/\/external.example\/\")\n\n assert fetcher.calls == 1\n assert result.http_status == 200\n assert repo.store\n\n\n@pytest.mark.asyncio\nasync def test_cache_hit_skips_fetch() -> None:\n fetcher = _StubFetcher(status=200)\n repo = FakeExternalCheckRepo()\n now = datetime(2026, 4, 21, 12, 0, 0)\n ttl = timedelta(hours=1)\n uc = CheckExternalLinkUseCase(\n fetcher=fetcher, checks=repo, normalizer=UrlNormalizer(),\n clock=FakeClock(now), cache_ttl=ttl,\n )\n await uc.execute(\"https:\/\/external.example\/\")\n\n uc_later = CheckExternalLinkUseCase(\n fetcher=fetcher, checks=repo, normalizer=UrlNormalizer(),\n clock=FakeClock(now + timedelta(minutes=10)), cache_ttl=ttl,\n )\n await uc_later.execute(\"https:\/\/external.example\/\")\n\n assert fetcher.calls == 1\n\n\n@pytest.mark.asyncio\nasync def test_cache_expired_refetches() -> None:\n fetcher = _StubFetcher(status=200)\n repo = FakeExternalCheckRepo()\n now = datetime(2026, 4, 21, 12, 0, 0)\n ttl = timedelta(hours=1)\n await CheckExternalLinkUseCase(\n fetcher=fetcher, checks=repo, normalizer=UrlNormalizer(),\n clock=FakeClock(now), cache_ttl=ttl,\n ).execute(\"https:\/\/external.example\/\")\n\n await CheckExternalLinkUseCase(\n fetcher=fetcher, checks=repo, normalizer=UrlNormalizer(),\n clock=FakeClock(now + timedelta(hours=2)), cache_ttl=ttl,\n ).execute(\"https:\/\/external.example\/\")\n\n assert fetcher.calls == 2\n"
}
}
{
"tool_response": {
"type": "create",
"filePath": "\/var\/www\/dev.campus.systemische-tools.de\/crawler\/tests\/unit\/application\/test_check_external_link.py",
"content": "from __future__ import annotations\n\nfrom collections.abc import Mapping\nfrom datetime import datetime, timedelta\n\nimport pytest\n\nfrom src.application.use_cases.check_external_link import CheckExternalLinkUseCase\nfrom src.domain.ports.http_fetcher_port import FetchResult, HttpFetcherPort\nfrom src.domain.services.url_normalizer import UrlNormalizer\nfrom src.domain.value_objects.normalized_url import NormalizedUrl\nfrom tests.unit.application.fakes import FakeClock, FakeExternalCheckRepo\n\n\nclass _StubFetcher(HttpFetcherPort):\n def __init__(self, status: int, error: str | None = None) -> None:\n self.calls = 0\n self._status = status\n self._error = error\n\n async def fetch(self, url: NormalizedUrl) -> FetchResult:\n self.calls += 1\n headers: Mapping[str, str] = {}\n return FetchResult(\n final_url=url.url,\n status_code=self._status,\n redirect_chain=(),\n headers=headers,\n body=b\"\",\n response_time_ms=12,\n ttfb_ms=None,\n error=self._error,\n )\n\n\n@pytest.mark.asyncio\nasync def test_first_check_fetches_and_persists() -> None:\n fetcher = _StubFetcher(status=200)\n repo = FakeExternalCheckRepo()\n now = datetime(2026, 4, 21, 12, 0, 0)\n uc = CheckExternalLinkUseCase(\n fetcher=fetcher, checks=repo, normalizer=UrlNormalizer(),\n clock=FakeClock(now), cache_ttl=timedelta(hours=1),\n )\n\n result = await uc.execute(\"https:\/\/external.example\/\")\n\n assert fetcher.calls == 1\n assert result.http_status == 200\n assert repo.store\n\n\n@pytest.mark.asyncio\nasync def test_cache_hit_skips_fetch() -> None:\n fetcher = _StubFetcher(status=200)\n repo = FakeExternalCheckRepo()\n now = datetime(2026, 4, 21, 12, 0, 0)\n ttl = timedelta(hours=1)\n uc = CheckExternalLinkUseCase(\n fetcher=fetcher, checks=repo, normalizer=UrlNormalizer(),\n clock=FakeClock(now), cache_ttl=ttl,\n )\n await uc.execute(\"https:\/\/external.example\/\")\n\n uc_later = CheckExternalLinkUseCase(\n fetcher=fetcher, checks=repo, normalizer=UrlNormalizer(),\n clock=FakeClock(now + timedelta(minutes=10)), cache_ttl=ttl,\n )\n await uc_later.execute(\"https:\/\/external.example\/\")\n\n assert fetcher.calls == 1\n\n\n@pytest.mark.asyncio\nasync def test_cache_expired_refetches() -> None:\n fetcher = _StubFetcher(status=200)\n repo = FakeExternalCheckRepo()\n now = datetime(2026, 4, 21, 12, 0, 0)\n ttl = timedelta(hours=1)\n await CheckExternalLinkUseCase(\n fetcher=fetcher, checks=repo, normalizer=UrlNormalizer(),\n clock=FakeClock(now), cache_ttl=ttl,\n ).execute(\"https:\/\/external.example\/\")\n\n await CheckExternalLinkUseCase(\n fetcher=fetcher, checks=repo, normalizer=UrlNormalizer(),\n clock=FakeClock(now + timedelta(hours=2)), cache_ttl=ttl,\n ).execute(\"https:\/\/external.example\/\")\n\n assert fetcher.calls == 2\n",
"structuredPatch": [],
"originalFile": null,
"userModified": false
}
}