from datetime import datetime, timedelta, timezone from types import SimpleNamespace from unittest.mock import AsyncMock import uuid import pytest from sqlalchemy.ext.asyncio import AsyncSession from common.database.model.models import places, sites, site_versions, site_search_status from crud import search_console_crud as store from services import search_console_service as service from services.search_console_client import SearchConsoleError from services.search_console_alerts import alert_reason, send_alert from services.search_console_settings import SearchConsoleSettings, belongs_to_property, load_settings NOW = datetime.now(timezone.utc) SETTINGS = SearchConsoleSettings("https://example.com/", "/test/key.json") def status(**changes): return SimpleNamespace(site_id=uuid.uuid4(), page_url="https://example.com/s/stay", property_url=SETTINGS.property_url, published_at=NOW - timedelta(days=8), sitemap_submitted_at=None, inspected_at=None, inspection=None, first_indexed_at=None, error_code=None, failures=0, alerted_at=None, next_check_at=NOW, **changes) @pytest.mark.parametrize("url,prop,expected", [ ("https://example.com/s/a", "https://example.com/", True), ("https://example.com.evil.test/s/a", "https://example.com/", False), ("https://sub.example.com/s/a", "sc-domain:example.com", True), ("https://notexample.com/s/a", "sc-domain:example.com", False), ("http://example.com/s/a", "sc-domain:example.com", False), ("https://example.com/s/a?secret=x", "sc-domain:example.com", False), ]) def test_property_scope(url, prop, expected): assert belongs_to_property(url, prop) is expected def test_disabled_needs_no_credentials(monkeypatch): monkeypatch.delenv("GSC_ENABLED", raising=False) assert load_settings() is None monkeypatch.setenv("GSC_ENABLED", "1") monkeypatch.delenv("GSC_PROPERTY_URL", raising=False) with pytest.raises(ValueError): load_settings() async def test_submit_and_inspect_are_separate(monkeypatch): row = status() client = SimpleNamespace(submit_sitemap=AsyncMock(), inspect_url=AsyncMock(return_value={"verdict": "NEUTRAL"})) monkeypatch.setattr(service, "read_sitemap", AsyncMock(return_value={row.page_url})) monkeypatch.setattr(service.site_payload, "publish_origin", lambda: "https://example.com") await service.check_site(client, row, SETTINGS, set()) assert row.sitemap_submitted_at assert row.inspected_at assert row.first_indexed_at is None assert row.next_check_at > NOW assert row.error_code is None async def test_missing_sitemap_url_retries_without_google_call(monkeypatch): row = status() client = SimpleNamespace(submit_sitemap=AsyncMock(), inspect_url=AsyncMock()) monkeypatch.setattr(service, "read_sitemap", AsyncMock(return_value=set())) await service.check_site(client, row, SETTINGS, set()) assert row.error_code == "URL_NOT_IN_SITEMAP" assert row.sitemap_submitted_at is None assert row.failures == 1 client.submit_sitemap.assert_not_awaited() client.inspect_url.assert_not_awaited() async def test_one_sitemap_submission_per_batch(monkeypatch): client = SimpleNamespace(submit_sitemap=AsyncMock()) monkeypatch.setattr(service, "read_sitemap", AsyncMock(return_value={status().page_url})) seen = set() await service.submit_sitemap(client, status(), seen) await service.submit_sitemap(client, status(), seen) client.submit_sitemap.assert_awaited_once() async def test_inspection_failure_preserves_previous_observation(): row = status() row.sitemap_submitted_at = NOW row.inspection = {"verdict": "PASS"} row.inspected_at = NOW - timedelta(days=1) client = SimpleNamespace(inspect_url=AsyncMock(side_effect=SearchConsoleError("HTTP_429"))) await service.check_site(client, row, SETTINGS, set()) assert row.inspection == {"verdict": "PASS"} assert row.inspected_at < NOW assert row.error_code == "HTTP_429" assert row.failures == 1 def test_first_indexed_is_observation_time_not_google_crawl_time(): row = status() service.record_inspection(row, {"verdict": "PASS", "lastCrawlTime": "2020-01-01T00:00:00Z"}, NOW) service.record_inspection(row, {"verdict": "PASS"}, NOW + timedelta(days=1)) assert row.first_indexed_at == NOW def test_alert_cooldown_and_backoff(): row = status() assert alert_reason(row, NOW, 7) row.alerted_at = NOW assert alert_reason(row, NOW, 7) is None for _ in range(30): service.record_error(row, "HTTP_503", NOW) assert row.next_check_at == NOW + timedelta(days=1) async def test_failed_alert_does_not_mark_delivered(monkeypatch): row = status() row.sitemap_submitted_at = NOW client = SimpleNamespace(inspect_url=AsyncMock(return_value={"verdict": "NEUTRAL"})) send = AsyncMock(return_value=False) monkeypatch.setattr(service, "send_alert", send) await service.check_site(client, row, SETTINGS, set()) send.assert_awaited_once() assert row.alerted_at is None async def test_empty_alert_url_never_calls_network(): assert not await send_alert("", "https://example.com/s/a", "delayed") async def test_disabled_cron_never_opens_database(monkeypatch): monkeypatch.delenv("GSC_ENABLED", raising=False) call = AsyncMock() monkeypatch.setattr(service.DB_SESSION_MNG, "execute_lambda_write", call) await service.run_scheduled_check() call.assert_not_awaited() async def seed_site(session, *, state=3): place = places(place_id=uuid.uuid4(), owner_user_id=uuid.uuid4(), name="숙소", category=1) site = sites(site_id=uuid.uuid4(), place_id=place.place_id, status=state, domain="stay", published_at=NOW) version = site_versions(site_version_id=uuid.uuid4(), site_id=site.site_id, version=1, build_status=2) site.current_version_id = version.site_version_id session.add_all([place, site, version]) await session.flush() return site, place, version async def test_registration_survives_restart_and_resets_on_publish(db_engine, monkeypatch): monkeypatch.setattr(service.site_payload, "publish_origin", lambda: "https://example.com") async with AsyncSession(db_engine, expire_on_commit=False) as session: site, _, _ = await seed_site(session) await service.register_publications(session, SETTINGS) await session.commit() row = await session.get(site_search_status, site.site_id) row.sitemap_submitted_at = NOW row.first_indexed_at = NOW await session.commit() await service.register_publications(session, SETTINGS) assert row.first_indexed_at == NOW site.published_at = NOW + timedelta(seconds=1) await session.commit() await service.register_publications(session, SETTINGS) await session.commit() await session.refresh(row) assert row.first_indexed_at is None assert row.sitemap_submitted_at is None async def test_unpublished_and_deleted_places_are_not_checked(db_engine, monkeypatch): monkeypatch.setattr(service.site_payload, "publish_origin", lambda: "https://example.com") async with AsyncSession(db_engine) as session: site, place, _ = await seed_site(session) await service.register_publications(session, SETTINGS) assert len(await store.due_sites(session, SETTINGS.property_url)) == 1 site.status = 5 await session.flush() assert not await store.due_sites(session, SETTINGS.property_url) site.status = 3 place.deleted = True await session.flush() assert not await store.due_sites(session, SETTINGS.property_url) async def test_batch_lock_excludes_second_process(db_engine): async with AsyncSession(db_engine) as first, AsyncSession(db_engine) as second: assert await store.lock_batch(first) assert not await store.lock_batch(second) await first.rollback() assert await store.lock_batch(second) async def test_full_batch_commits_and_does_not_repeat_same_day(db_engine, monkeypatch): monkeypatch.setattr(service.site_payload, "publish_origin", lambda: "https://example.com") monkeypatch.setattr(service, "read_sitemap", AsyncMock(return_value={status().page_url})) client = SimpleNamespace(submit_sitemap=AsyncMock(), inspect_url=AsyncMock(return_value={"verdict": "PASS"})) class FakeClient: def __init__(self, *_): pass async def __aenter__(self): return client async def __aexit__(self, *_): pass monkeypatch.setattr(service, "SearchConsoleClient", FakeClient) async with AsyncSession(db_engine) as session: site, _, _ = await seed_site(session) sid = site.site_id await session.commit() await service.run_batch(session, SETTINGS) await session.commit() async with AsyncSession(db_engine) as session: row = await session.get(site_search_status, sid) assert row.sitemap_submitted_at assert row.first_indexed_at await service.run_batch(session, SETTINGS) await session.commit() client.submit_sitemap.assert_awaited_once() client.inspect_url.assert_awaited_once() async def test_cron_failure_is_isolated(monkeypatch): monkeypatch.setattr(service, "load_settings", lambda: SETTINGS) monkeypatch.setattr(service.DB_SESSION_MNG, "execute_lambda_write", AsyncMock(side_effect=RuntimeError("db"))) await service.run_scheduled_check() async def test_sitemap_http_parser(monkeypatch): import httpx real_client = httpx.AsyncClient transport = httpx.MockTransport(lambda request: httpx.Response(200, text=''' https://example.com/s/stay''')) monkeypatch.setattr(service.httpx, "AsyncClient", lambda **kw: real_client(transport=transport, **kw)) assert await service.read_sitemap("https://example.com/sitemap.xml") == {status().page_url} async def test_sitemap_redirect_is_not_followed(monkeypatch): import httpx real_client = httpx.AsyncClient transport = httpx.MockTransport(lambda request: httpx.Response(302, headers={"location": "http://internal/"})) monkeypatch.setattr(service.httpx, "AsyncClient", lambda **kw: real_client(transport=transport, **kw)) with pytest.raises(SearchConsoleError, match="SITEMAP_UNAVAILABLE"): await service.read_sitemap("https://example.com/sitemap.xml") @pytest.mark.parametrize("enabled,count", [("0", 0), ("1", 1)]) def test_existing_scheduler_registers_optional_job(monkeypatch, enabled, count): import scheduler instance = SimpleNamespace(add_job=lambda *a, **kw: jobs.append((a, kw)), start=lambda: None, get_jobs=lambda: jobs) jobs = [] monkeypatch.setattr(scheduler, "_scheduler", None) monkeypatch.setattr(scheduler, "AsyncIOScheduler", lambda **kw: instance) monkeypatch.setenv("SCHEDULER_ENABLED", "1") monkeypatch.setenv("GSC_ENABLED", enabled) scheduler.start_scheduler() assert len(jobs) == count if jobs: assert jobs[0][1]["minutes"] == 10 assert jobs[0][1]["max_instances"] == 1 def test_migration_matches_fresh_database_schema(): from pathlib import Path root = Path(__file__).resolve().parents[3] migration = (root / "postgres-init/migrations/0014_search_console.sql").read_text() ddl = migration[migration.index("CREATE TABLE"):].strip() assert ddl in (root / "postgres-init/init-data/init.sql").read_text()