gti
d21a9d1140
git-subtree-dir: fiscal git-subtree-mainline:d95165fc8fgit-subtree-split:102ba84500
63 lines
3.2 KiB
Python
63 lines
3.2 KiB
Python
from fastapi.testclient import TestClient
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
from central.app import api as api_module
|
|
from central.app.main import create_app
|
|
from central.app.models import Base
|
|
|
|
HEADERS = {"Authorization": "Bearer test-token"}
|
|
PAYLOAD = {"current_ncm": "22030000", "description": "CERVEJA LATA"}
|
|
|
|
|
|
def make_client(monkeypatch, calls, engine_version="0.3.0", datasets=None):
|
|
versions = datasets if datasets is not None else {"ncm": "aaa"}
|
|
|
|
class FakeClassifier:
|
|
def __init__(self, repository): pass
|
|
|
|
def classify(self, product):
|
|
calls.append(product)
|
|
return {"status": "EXACT_NCM", "engine_version": api_module.ENGINE_VERSION, "decision_version": "x", "dataset_versions": dict(versions), "classification": {"ncm": product["current_ncm"]}}
|
|
|
|
monkeypatch.setattr(api_module, "Classifier", FakeClassifier)
|
|
monkeypatch.setattr(api_module, "ENGINE_VERSION", engine_version)
|
|
monkeypatch.setattr(api_module.DatasetService, "dataset_versions", lambda self: dict(versions))
|
|
engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
|
|
Base.metadata.create_all(engine)
|
|
return TestClient(create_app(session_factory=lambda: Session(engine), bootstrap_token="test-token")), versions
|
|
|
|
|
|
def test_cache_hit_does_not_run_the_classifier(monkeypatch):
|
|
calls = []
|
|
client, _ = make_client(monkeypatch, calls)
|
|
first = client.post("/v1/classify", json=PAYLOAD, headers=HEADERS).json()
|
|
assert first["provenance"]["cache_hit"] is False and len(calls) == 1
|
|
for _ in range(3):
|
|
again = client.post("/v1/classify", json=PAYLOAD, headers=HEADERS).json()
|
|
assert again["provenance"]["cache_hit"] is True
|
|
assert {k: v for k, v in again.items() if k != "provenance"} == {k: v for k, v in first.items() if k != "provenance"} # same fiscal result
|
|
assert len(calls) == 1 # CLASSIFIER_EXECUTIONS_ON_HIT = 0
|
|
|
|
|
|
def test_batch_hits_skip_the_classifier_and_new_inputs_miss(monkeypatch):
|
|
calls = []
|
|
client, _ = make_client(monkeypatch, calls)
|
|
products = [PAYLOAD, {"current_ncm": "19022000", "description": "MASSA"}]
|
|
client.post("/v1/classify/batch", json={"products": products}, headers=HEADERS)
|
|
assert len(calls) == 2
|
|
res = client.post("/v1/classify/batch", json={"products": products + [{"current_ncm": "20099000", "description": "SUCO"}]}, headers=HEADERS).json()["results"]
|
|
assert [r["provenance"]["cache_hit"] for r in res] == [True, True, False] and len(calls) == 3
|
|
|
|
|
|
def test_engine_or_dataset_version_change_invalidates_naturally(monkeypatch):
|
|
calls = []
|
|
client, versions = make_client(monkeypatch, calls)
|
|
client.post("/v1/classify", json=PAYLOAD, headers=HEADERS)
|
|
versions["ncm"] = "bbb" # a new ACTIVE dataset version changes dataset_versions()
|
|
assert client.post("/v1/classify", json=PAYLOAD, headers=HEADERS).json()["provenance"]["cache_hit"] is False
|
|
monkeypatch.setattr(api_module, "ENGINE_VERSION", "0.4.0")
|
|
assert client.post("/v1/classify", json=PAYLOAD, headers=HEADERS).json()["provenance"]["cache_hit"] is False
|
|
assert len(calls) == 3
|