132 lines
2.6 KiB
Python
132 lines
2.6 KiB
Python
from fastapi.testclient import TestClient
|
|
|
|
from core.api.main import app
|
|
|
|
|
|
client = TestClient(app)
|
|
|
|
|
|
def valid_payload() -> dict:
|
|
return {
|
|
"raw_id": "raw-test-001",
|
|
"src_ip": "10.240.12.1",
|
|
"device_id": "310410998877665",
|
|
"caller_num": "0019875551234",
|
|
"svc": "4G_DATA",
|
|
"up_bytes": 1048576,
|
|
"down_bytes": 4194304,
|
|
"rat": 6,
|
|
"lac": "4021",
|
|
"ci": "8812",
|
|
"epoch_start": 1789900000,
|
|
"epoch_end": 1789900300,
|
|
}
|
|
|
|
|
|
def test_health_endpoint() -> None:
|
|
response = client.get("/health")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == {
|
|
"status": "ok",
|
|
}
|
|
|
|
|
|
def test_valid_cdr_returns_202() -> None:
|
|
payload = valid_payload()
|
|
|
|
response = client.post(
|
|
"/api/v1/cdr/raw",
|
|
json=payload,
|
|
)
|
|
|
|
assert response.status_code == 202
|
|
|
|
assert response.json() == {
|
|
"status": "accepted",
|
|
"raw_id": "raw-test-001",
|
|
}
|
|
|
|
|
|
def test_missing_required_field_returns_400() -> None:
|
|
payload = valid_payload()
|
|
del payload["caller_num"]
|
|
|
|
response = client.post(
|
|
"/api/v1/cdr/raw",
|
|
json=payload,
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
|
|
body = response.json()
|
|
|
|
assert body["error"] == "missing_required_field"
|
|
assert body["failed_field"] == "caller_num"
|
|
|
|
|
|
def test_invalid_rat_returns_400() -> None:
|
|
payload = valid_payload()
|
|
payload["rat"] = 99
|
|
|
|
response = client.post(
|
|
"/api/v1/cdr/raw",
|
|
json=payload,
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
|
|
body = response.json()
|
|
|
|
assert body["error"] == "invalid_rat_value"
|
|
assert body["failed_field"] == "rat"
|
|
|
|
|
|
def test_invalid_epoch_returns_400() -> None:
|
|
payload = valid_payload()
|
|
|
|
payload["epoch_start"] = 200
|
|
payload["epoch_end"] = 100
|
|
|
|
response = client.post(
|
|
"/api/v1/cdr/raw",
|
|
json=payload,
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
|
|
body = response.json()
|
|
|
|
assert body["error"] == "invalid_epoch"
|
|
|
|
def test_malformed_json_returns_400() -> None:
|
|
response = client.post(
|
|
"/api/v1/cdr/raw",
|
|
content='{"raw_id": invalid}',
|
|
headers={
|
|
"Content-Type": "application/json",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
|
|
body = response.json()
|
|
|
|
assert body["error"] == "invalid_json"
|
|
|
|
def test_non_object_json_returns_400() -> None:
|
|
response = client.post(
|
|
"/api/v1/cdr/raw",
|
|
json=[
|
|
"not",
|
|
"a",
|
|
"cdr",
|
|
],
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
|
|
body = response.json()
|
|
|
|
assert body["error"] == "invalid_payload"
|