mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-06-08 11:17:27 +00:00
96 lines
2.8 KiB
Python
96 lines
2.8 KiB
Python
import uuid
|
|
from datetime import datetime
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from backend.app.db.connection import get_session
|
|
from backend.app.dependencies import validate_token
|
|
from backend.app.db.models.tasks import SourceEnum, Task
|
|
from backend.app.tasks.router import router
|
|
|
|
|
|
def test_get_task_by_source(db_session: Session) -> None:
|
|
# arrange
|
|
current_categorisation_task = Task(
|
|
id=uuid.uuid4(),
|
|
task_source="",
|
|
job_started=datetime(2026, 1, 1, 9, 0, 0),
|
|
job_completed=None,
|
|
status="in progress",
|
|
service="plan_categorisation",
|
|
updated_at=datetime(2026, 1, 1, 9, 0, 0),
|
|
source=SourceEnum.PORTFOLIO,
|
|
source_id="100",
|
|
)
|
|
|
|
previous_categorisation_task = Task(
|
|
id=uuid.uuid4(),
|
|
task_source="",
|
|
job_started=datetime(2025, 12, 31, 9, 0, 0),
|
|
job_completed=datetime(2025, 12, 31, 9, 10, 0),
|
|
status="complete",
|
|
service="plan_categorisation",
|
|
updated_at=datetime(2025, 12, 31, 9, 10, 0),
|
|
source=SourceEnum.PORTFOLIO,
|
|
source_id="100",
|
|
)
|
|
|
|
other_portfolio_categorisation_task = Task(
|
|
id=uuid.uuid4(),
|
|
task_source="",
|
|
job_started=datetime(2026, 1, 1, 9, 0, 0),
|
|
job_completed=None,
|
|
status="in progress",
|
|
service="plan_categorisation",
|
|
updated_at=datetime(2026, 1, 1, 9, 0, 0),
|
|
source=SourceEnum.PORTFOLIO,
|
|
source_id="101",
|
|
)
|
|
|
|
engine_task = Task(
|
|
id=uuid.uuid4(),
|
|
task_source="",
|
|
job_started=datetime(2026, 1, 1, 9, 0, 0),
|
|
job_completed=None,
|
|
status="in progress",
|
|
service="plan_engine",
|
|
updated_at=datetime(2026, 1, 1, 9, 0, 0),
|
|
source=SourceEnum.PORTFOLIO,
|
|
source_id="100",
|
|
)
|
|
|
|
db_session.add_all(
|
|
[
|
|
current_categorisation_task,
|
|
previous_categorisation_task,
|
|
other_portfolio_categorisation_task,
|
|
engine_task,
|
|
]
|
|
)
|
|
db_session.commit()
|
|
# db_session.flush()
|
|
|
|
# debug: confirm data is visible in this session
|
|
all_tasks = db_session.execute(select(Task)).scalars().all()
|
|
print(f"Tasks in db: {[(t.service, t.source_id, t.source) for t in all_tasks]}")
|
|
|
|
# act
|
|
test_app = FastAPI()
|
|
test_app.include_router(router)
|
|
|
|
def override_get_session():
|
|
yield db_session
|
|
|
|
test_app.dependency_overrides[get_session] = override_get_session
|
|
test_app.dependency_overrides[validate_token] = lambda: None
|
|
|
|
client = TestClient(test_app)
|
|
response = client.get("/tasks/by-source/portfolio_id/100/plan_categorisation")
|
|
|
|
test_app.dependency_overrides.clear()
|
|
|
|
# assert
|
|
assert response.status_code == 200
|
|
assert response.json()["task"]["id"] == str(current_categorisation_task.id)
|