97 lines
2.7 KiB
Python
97 lines
2.7 KiB
Python
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy.orm import Session
|
|
import xmltodict
|
|
import pytest
|
|
from src.main import app, get_db
|
|
from src import crud, models, schemas
|
|
from src.database import Base, engine, SessionLocal
|
|
|
|
# Dependency override
|
|
def override_get_db():
|
|
try:
|
|
db = SessionLocal()
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
|
|
app.dependency_overrides[get_db] = override_get_db
|
|
|
|
@pytest.fixture(scope="module")
|
|
def client():
|
|
with TestClient(app) as c:
|
|
yield c
|
|
|
|
@pytest.fixture(scope="function")
|
|
def db_session():
|
|
Base.metadata.create_all(bind=engine)
|
|
db = SessionLocal()
|
|
try:
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
Base.metadata.drop_all(bind=engine)
|
|
|
|
def test_openbin_outbound_returns_task(client, db_session):
|
|
"""
|
|
Tests that OpenBin on an OUTBOUND port returns the bin associated with a queued task.
|
|
"""
|
|
# 1. Setup: Create Port (OUTBOUND), Bin, and TaskGroup with Task
|
|
port_id = "P-OUT-TASK-01"
|
|
bin_id = "BIN-TASK-01"
|
|
task_group_id = "TG-TASK-01"
|
|
task_id = "T-TASK-01"
|
|
|
|
# Create Port
|
|
crud.create_port(db_session, schemas.PortCreate(id=port_id, port_type="OUTBOUND"))
|
|
|
|
# Create Bin
|
|
crud.create_bin(db_session, schemas.BinCreate(id=bin_id, content_code=100))
|
|
|
|
# Create TaskGroup with Task (Unassigned to port initially)
|
|
# We use the XML API to create it to simulate real usage
|
|
xml_create_tg = f"""
|
|
<AutoStore>
|
|
<create_taskgroup>
|
|
<params>
|
|
<taskgroup_id>{task_group_id}</taskgroup_id>
|
|
<tasks>
|
|
<task>
|
|
<task_id>{task_id}</task_id>
|
|
<bin_id>{bin_id}</bin_id>
|
|
</task>
|
|
</tasks>
|
|
</params>
|
|
</create_taskgroup>
|
|
</AutoStore>
|
|
"""
|
|
headers = {"Content-Type": "application/xml"}
|
|
response = client.post("/api", content=xml_create_tg, headers=headers)
|
|
assert response.status_code == 200
|
|
|
|
# 2. Act: Call OpenBin for the Port
|
|
xml_open_bin = f"""
|
|
<AutoStore>
|
|
<OpenBin>
|
|
<Port>{port_id}</Port>
|
|
</OpenBin>
|
|
</AutoStore>
|
|
"""
|
|
response = client.post("/api", content=xml_open_bin, headers=headers)
|
|
|
|
# 3. Assert: Verify response contains the task info
|
|
assert response.status_code == 200
|
|
data = xmltodict.parse(response.content)
|
|
|
|
# Check structure
|
|
assert "response" in data
|
|
assert "params" in data["response"]
|
|
params = data["response"]["params"]
|
|
|
|
assert params["bin_id"] == bin_id
|
|
assert params["port_id"] == port_id
|
|
assert params["task_id"] == task_id
|
|
assert params["taskgroup_id"] == task_group_id
|
|
assert params["success"] == "true"
|
|
|