mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-07-12 13:29:04 +00:00
119ff3740c
@task_handler never built or passed cloud_logs_url, so every app using it (incl. modelling_e2e) ran run_subtask with the None default and the CloudWatch deep-link was never saved onto the SubTask. @subtask_handler did this correctly. Extract the URL builder into a shared utilities/aws_lambda/cloud_logs.py (public cloudwatch_url()), use it from both handlers, and pass the URL into run_subtask from @task_handler. Add regression tests. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
104 lines
3.1 KiB
Python
104 lines
3.1 KiB
Python
from collections.abc import Generator, Iterator
|
|
from contextlib import contextmanager
|
|
from dataclasses import dataclass
|
|
from typing import Any
|
|
from uuid import UUID
|
|
|
|
import pytest
|
|
from sqlalchemy import Engine
|
|
from sqlmodel import Session
|
|
|
|
from domain.tasks.tasks import Source
|
|
from orchestration.task_orchestrator import TaskOrchestrator
|
|
from repositories.tasks.subtask_postgres_repository import SubTaskPostgresRepository
|
|
from repositories.tasks.task_postgres_repository import TaskPostgresRepository
|
|
from utilities.aws_lambda.task_handler import task_handler
|
|
|
|
|
|
@dataclass
|
|
class Harness:
|
|
orchestrator: TaskOrchestrator
|
|
tasks: TaskPostgresRepository
|
|
subtasks: SubTaskPostgresRepository
|
|
|
|
@contextmanager
|
|
def factory(self) -> Generator[TaskOrchestrator, None, None]:
|
|
yield self.orchestrator
|
|
|
|
|
|
@pytest.fixture
|
|
def harness(db_engine: Engine) -> Iterator[Harness]:
|
|
with Session(db_engine) as session:
|
|
tasks = TaskPostgresRepository(session=session)
|
|
subtasks = SubTaskPostgresRepository(session=session)
|
|
yield Harness(
|
|
orchestrator=TaskOrchestrator(task_repo=tasks, subtask_repo=subtasks),
|
|
tasks=tasks,
|
|
subtasks=subtasks,
|
|
)
|
|
|
|
|
|
def _direct_event(property_id: str) -> dict[str, Any]:
|
|
return {"property_id": property_id}
|
|
|
|
|
|
def test_task_handler_records_cloudwatch_url_on_subtask(
|
|
harness: Harness, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
# arrange
|
|
monkeypatch.setenv("AWS_REGION", "eu-west-2")
|
|
monkeypatch.setenv(
|
|
"AWS_LAMBDA_LOG_GROUP_NAME", "/aws/lambda/modelling-e2e"
|
|
)
|
|
monkeypatch.setenv(
|
|
"AWS_LAMBDA_LOG_STREAM_NAME", "2026/05/20/[$LATEST]abc123"
|
|
)
|
|
|
|
@task_handler(
|
|
task_source="modelling_e2e",
|
|
source=Source.PROPERTY,
|
|
orchestrator_cm=harness.factory,
|
|
)
|
|
def handler(body: dict[str, Any], context: Any) -> None:
|
|
return None
|
|
|
|
# act
|
|
result = handler(_direct_event("prop-1"), context=None)
|
|
|
|
# assert
|
|
subtask_id = result[0]["subtask_id"]
|
|
saved_url = harness.subtasks.get(UUID(subtask_id)).cloud_logs_url
|
|
assert saved_url is not None
|
|
assert saved_url.startswith(
|
|
"https://eu-west-2.console.aws.amazon.com/cloudwatch/home"
|
|
)
|
|
# Log group / stream are console-encoded ("/" -> "$252F").
|
|
assert "$252Faws$252Flambda$252Fmodelling-e2e" in saved_url
|
|
assert "$255B$2524LATEST$255D" in saved_url
|
|
|
|
|
|
def test_task_handler_leaves_cloudwatch_url_unset_outside_lambda(
|
|
harness: Harness, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
# arrange
|
|
for var in (
|
|
"AWS_REGION",
|
|
"AWS_LAMBDA_LOG_GROUP_NAME",
|
|
"AWS_LAMBDA_LOG_STREAM_NAME",
|
|
):
|
|
monkeypatch.delenv(var, raising=False)
|
|
|
|
@task_handler(
|
|
task_source="modelling_e2e",
|
|
source=Source.PROPERTY,
|
|
orchestrator_cm=harness.factory,
|
|
)
|
|
def handler(body: dict[str, Any], context: Any) -> None:
|
|
return None
|
|
|
|
# act
|
|
result = handler(_direct_event("prop-1"), context=None)
|
|
|
|
# assert
|
|
subtask_id = result[0]["subtask_id"]
|
|
assert harness.subtasks.get(UUID(subtask_id)).cloud_logs_url is None
|