GitHub Copilot
8e0c13ae17
- Add OFF mode to PortType enum in schemas - Add set_port_mode CRUD function with mode validation - Implement SetPortMode XML command handler with comprehensive validation - Add 6 comprehensive tests covering all scenarios: success, OFF mode, invalid mode, port not found, missing port/mode - All 39 tests passing with no regressions
960 lines
34 KiB
Python
960 lines
34 KiB
Python
import os
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
from src.main import app, get_db
|
|
from src.database import Base
|
|
import xmltodict
|
|
from src import crud, schemas
|
|
|
|
# --- Test Database Setup ---
|
|
TEST_DATABASE_URL = "sqlite:///./test.db"
|
|
|
|
engine = create_engine(
|
|
TEST_DATABASE_URL, connect_args={"check_same_thread": False}
|
|
)
|
|
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
|
|
|
|
|
# --- Fixtures ---
|
|
|
|
@pytest.fixture(scope="function")
|
|
def db_session():
|
|
"""
|
|
Creates a clean database and a new session for each test function.
|
|
This ensures test isolation by creating and dropping tables for each test.
|
|
"""
|
|
Base.metadata.create_all(bind=engine)
|
|
db = TestingSessionLocal()
|
|
try:
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
Base.metadata.drop_all(bind=engine)
|
|
|
|
@pytest.fixture(scope="function")
|
|
def client(db_session):
|
|
"""
|
|
Creates a test client with the database dependency overridden for each test.
|
|
This fixture receives the `db_session` fixture to ensure the app and tests
|
|
use the same, isolated database session.
|
|
"""
|
|
def override_get_db():
|
|
yield db_session
|
|
|
|
app.dependency_overrides[get_db] = override_get_db
|
|
yield TestClient(app)
|
|
app.dependency_overrides.clear()
|
|
|
|
|
|
# --- Tests ---
|
|
|
|
def test_health_check(client):
|
|
"""Tests the health check endpoint."""
|
|
response = client.get("/health")
|
|
assert response.status_code == 200
|
|
assert response.json() == {"status": "ok"}
|
|
|
|
def test_create_and_read_bin(client):
|
|
"""Tests creating a new BIN and then reading it."""
|
|
bin_id = "A-TEST-001"
|
|
response_create = client.post("/bins/", json={"id": bin_id, "content_code": 1})
|
|
assert response_create.status_code == 200
|
|
data = response_create.json()
|
|
assert data["id"] == bin_id
|
|
|
|
response_read = client.get(f"/bins/{bin_id}")
|
|
assert response_read.status_code == 200
|
|
assert response_read.json() == data
|
|
|
|
def test_create_existing_bin_fails(client):
|
|
"""Tests that creating a BIN that already exists fails with a 400 error."""
|
|
bin_id = "A-TEST-002"
|
|
client.post("/bins/", json={"id": bin_id, "content_code": 1}) # First time should be OK
|
|
response = client.post("/bins/", json={"id": bin_id, "content_code": 1}) # Second time should fail
|
|
assert response.status_code == 400
|
|
assert response.json() == {"detail": "Bin already registered"}
|
|
|
|
def test_read_nonexistent_bin(client):
|
|
"""Tests that reading a non-existent BIN returns a 404 error."""
|
|
response = client.get("/bins/NON-EXISTENT-BIN")
|
|
assert response.status_code == 404
|
|
|
|
def test_read_all_bins(client):
|
|
"""Tests reading a list of all BINs."""
|
|
client.post("/bins/", json={"id": "A-TEST-100", "content_code": 1})
|
|
client.post("/bins/", json={"id": "A-TEST-101", "content_code": 1})
|
|
|
|
response = client.get("/bins/")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert len(data) == 2
|
|
|
|
def test_delete_bin(client):
|
|
"""Tests deleting a BIN."""
|
|
bin_id = "A-TEST-200"
|
|
client.post("/bins/", json={"id": bin_id, "content_code": 1})
|
|
|
|
delete_response = client.delete(f"/bins/{bin_id}")
|
|
assert delete_response.status_code == 200
|
|
assert delete_response.json()["id"] == bin_id
|
|
|
|
# Verify it's gone
|
|
get_response = client.get(f"/bins/{bin_id}")
|
|
assert get_response.status_code == 404
|
|
|
|
def test_delete_non_existent_bin(client):
|
|
"""Tests that deleting a non-existent BIN returns a 404 error."""
|
|
response = client.delete("/bins/I-DO-NOT-EXIST")
|
|
assert response.status_code == 404
|
|
|
|
def test_generate_bins(client):
|
|
"""Tests generating a number of bins."""
|
|
generation_request = {"count": 50, "prefix": "GEN"}
|
|
response = client.post("/bins/generate", json=generation_request)
|
|
assert response.status_code == 201
|
|
data = response.json()
|
|
assert data["message"] == "50 bins generated successfully."
|
|
assert data["count"] == 50
|
|
|
|
# Verify that the bins were actually created
|
|
response_get = client.get("/bins/")
|
|
assert response_get.status_code == 200
|
|
assert len(response_get.json()) == 50
|
|
|
|
# Check if one of the generated bins exists
|
|
response_get_one = client.get("/bins/GEN-00001")
|
|
assert response_get_one.status_code == 200
|
|
|
|
|
|
# --- Port Tests ---
|
|
|
|
def test_create_and_read_port(client):
|
|
"""Tests creating a new port and then reading it."""
|
|
port_id = "P01"
|
|
port_type = "INBOUND"
|
|
response_create = client.post("/ports/", json={"id": port_id, "port_type": port_type})
|
|
assert response_create.status_code == 201
|
|
data = response_create.json()
|
|
assert data["id"] == port_id
|
|
assert data["port_type"] == port_type
|
|
assert data["status"] == "FREE"
|
|
|
|
response_read = client.get(f"/ports/{port_id}")
|
|
assert response_read.status_code == 200
|
|
assert response_read.json() == data
|
|
|
|
def test_create_existing_port_fails(client):
|
|
"""Tests that creating a port that already exists fails."""
|
|
port_id = "P02"
|
|
client.post("/ports/", json={"id": port_id, "port_type": "OUTBOUND"})
|
|
response = client.post("/ports/", json={"id": port_id, "port_type": "OUTBOUND"})
|
|
assert response.status_code == 400
|
|
assert response.json() == {"detail": "Port already registered"}
|
|
|
|
def test_read_nonexistent_port(client):
|
|
"""Tests that reading a non-existent port returns a 404 error."""
|
|
response = client.get("/ports/P99")
|
|
assert response.status_code == 404
|
|
|
|
def test_read_all_ports(client):
|
|
"""Tests reading a list of all ports."""
|
|
client.post("/ports/", json={"id": "P03", "port_type": "INBOUND"})
|
|
client.post("/ports/", json={"id": "P04", "port_type": "OUTBOUND"})
|
|
|
|
response = client.get("/ports/")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert len(data) >= 2 # Use >= in case other tests created ports
|
|
|
|
def test_port_arrive(client):
|
|
"""Tests the arrive logic for an inbound port."""
|
|
# 1. Create a BIN and an INBOUND port
|
|
bin_id = "BIN-ARRIVE-01"
|
|
port_id = "P-IN-01"
|
|
client.post("/bins/", json={"id": bin_id, "content_code": 1})
|
|
client.post("/ports/", json={"id": port_id, "port_type": "INBOUND"})
|
|
|
|
# 2. Signal arrival
|
|
response = client.post(f"/ports/{port_id}/arrive", json={"bin_id": bin_id})
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["status"] == "OCCUPIED"
|
|
assert data["bin_id"] == bin_id
|
|
|
|
# 3. Check error cases
|
|
# Port not free
|
|
response_not_free = client.post(f"/ports/{port_id}/arrive", json={"bin_id": bin_id})
|
|
assert response_not_free.status_code == 400
|
|
# Wrong port type
|
|
outbound_port_id = "P-OUT-01"
|
|
client.post("/ports/", json={"id": outbound_port_id, "port_type": "OUTBOUND"})
|
|
response_wrong_type = client.post(f"/ports/{outbound_port_id}/arrive", json={"bin_id": bin_id})
|
|
assert response_wrong_type.status_code == 400
|
|
# Bin not found at a different, free port
|
|
free_port_id = "P-IN-FREE"
|
|
client.post("/ports/", json={"id": free_port_id, "port_type": "INBOUND"})
|
|
response_bin_not_found = client.post(f"/ports/{free_port_id}/arrive", json={"bin_id": "FAKE-BIN"})
|
|
assert response_bin_not_found.status_code == 404
|
|
|
|
|
|
def test_port_depart(client, db_session):
|
|
"""Tests the depart logic for an outbound port."""
|
|
# 1. Create a BIN and an OUTBOUND port
|
|
bin_id = "BIN-DEPART-01"
|
|
port_id = "P-OUT-02"
|
|
crud.create_bin(db_session, schemas.BinCreate(id=bin_id, content_code=1))
|
|
port = crud.create_port(db_session, schemas.PortCreate(id=port_id, port_type="OUTBOUND"))
|
|
|
|
# Manually occupy the port for the test scenario
|
|
port.status = "OCCUPIED"
|
|
port.bin_id = bin_id
|
|
db_session.add(port)
|
|
db_session.commit()
|
|
db_session.refresh(port)
|
|
|
|
# 2. Signal departure
|
|
response = client.post(f"/ports/{port_id}/depart")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["status"] == "FREE"
|
|
assert data["bin_id"] is None
|
|
|
|
# 3. Check error cases
|
|
# Port not occupied
|
|
response_not_occupied = client.post(f"/ports/{port_id}/depart")
|
|
assert response_not_occupied.status_code == 400
|
|
# Wrong port type
|
|
inbound_port_id = "P-IN-02"
|
|
client.post("/ports/", json={"id": inbound_port_id, "port_type": "INBOUND"})
|
|
response_wrong_type = client.post(f"/ports/{inbound_port_id}/depart")
|
|
assert response_wrong_type.status_code == 400
|
|
|
|
|
|
# --- XML Interface Tests ---
|
|
|
|
def test_create_taskgroup_xml(client, db_session):
|
|
"""Tests creating a task group via the XML interface."""
|
|
# 1. Create necessary Bins and Port in the database first
|
|
port_id = "P-XML-01"
|
|
bin_id_1 = "BIN-XML-01"
|
|
bin_id_2 = "BIN-XML-02"
|
|
crud.create_port(db_session, schemas.PortCreate(id=port_id, port_type="OUTBOUND"))
|
|
crud.create_bin(db_session, schemas.BinCreate(id=bin_id_1, content_code=1))
|
|
crud.create_bin(db_session, schemas.BinCreate(id=bin_id_2, content_code=1))
|
|
db_session.commit()
|
|
|
|
# 2. Construct the XML request body
|
|
xml_request = f"""
|
|
<AutoStore>
|
|
<CreateTaskgroup>
|
|
<Taskgroup>TG-XML-01</Taskgroup>
|
|
<Port>{port_id}</Port>
|
|
<Tasks>
|
|
<Task>
|
|
<Task>TASK-XML-01</Task>
|
|
<Bin>{bin_id_1}</Bin>
|
|
</Task>
|
|
<Task>
|
|
<Task>TASK-XML-02</Task>
|
|
<Bin>{bin_id_2}</Bin>
|
|
</Task>
|
|
</Tasks>
|
|
</CreateTaskgroup>
|
|
</AutoStore>
|
|
"""
|
|
|
|
# 3. Send the request to the /api endpoint
|
|
headers = {"Content-Type": "application/xml"}
|
|
response = client.post("/api", content=xml_request, headers=headers)
|
|
|
|
# 4. Assert the response
|
|
assert response.status_code == 200
|
|
assert "application/xml" in response.headers["content-type"]
|
|
|
|
# 5. Parse the XML response and validate its content
|
|
response_data = xmltodict.parse(response.content)
|
|
assert "AutoStore" in response_data
|
|
assert "CreateTaskgroup" in response_data["AutoStore"]
|
|
assert "ExecutionTime" in response_data["AutoStore"]["CreateTaskgroup"]
|
|
|
|
# 6. Verify that the data was actually created in the database
|
|
db_task_group = crud.get_task_group(db_session, "TG-XML-01")
|
|
assert db_task_group is not None
|
|
assert db_task_group.id == "TG-XML-01"
|
|
assert db_task_group.port_id == port_id
|
|
assert len(db_task_group.tasks) == 2
|
|
assert db_task_group.tasks[0].id == "TASK-XML-01"
|
|
assert db_task_group.tasks[1].bin_id == bin_id_2
|
|
assert db_task_group.status == "queued"
|
|
assert db_task_group.tasks[0].status == "queued"
|
|
|
|
|
|
def test_create_task_group_fails_if_port_not_found_xml(client, db_session):
|
|
"""Tests that creating a task group via XML fails if the port does not exist."""
|
|
# 1. Create Bins
|
|
crud.create_bin(db_session, schemas.BinCreate(id="BIN-TG-F-01", content_code=1))
|
|
db_session.commit()
|
|
|
|
# 2. Construct the XML request body with a non-existent port
|
|
xml_request = """
|
|
<AutoStore>
|
|
<CreateTaskgroup>
|
|
<Taskgroup>TG-FAIL-XML-01</Taskgroup>
|
|
<Port>NON-EXISTENT-PORT</Port>
|
|
<Tasks>
|
|
<Task>
|
|
<Task>T-FAIL-XML-01</Task>
|
|
<Bin>BIN-TG-F-01</Bin>
|
|
</Task>
|
|
</Tasks>
|
|
</CreateTaskgroup>
|
|
</AutoStore>
|
|
"""
|
|
|
|
# 3. Send the request
|
|
headers = {"Content-Type": "application/xml"}
|
|
response = client.post("/api", content=xml_request, headers=headers)
|
|
|
|
# 4. Assert the response
|
|
assert response.status_code == 404
|
|
response_data = xmltodict.parse(response.content)
|
|
assert "Error" in response_data["AutoStore"]
|
|
assert "Port not found" in response_data["AutoStore"]["Error"]["Description"]
|
|
|
|
|
|
def test_create_task_group_fails_if_bin_not_found_xml(client, db_session):
|
|
"""Tests that creating a task group via XML fails if a bin does not exist."""
|
|
# 1. Create Port
|
|
crud.create_port(db_session, schemas.PortCreate(id="P-TG-F-01", port_type="OUTBOUND"))
|
|
db_session.commit()
|
|
|
|
# 2. Construct the XML request body with a non-existent bin
|
|
xml_request = """
|
|
<AutoStore>
|
|
<CreateTaskgroup>
|
|
<Taskgroup>TG-FAIL-XML-02</Taskgroup>
|
|
<Port>P-TG-F-01</Port>
|
|
<Tasks>
|
|
<Task>
|
|
<Task>T-FAIL-XML-02</Task>
|
|
<Bin>NON-EXISTENT-BIN</Bin>
|
|
</Task>
|
|
</Tasks>
|
|
</CreateTaskgroup>
|
|
</AutoStore>
|
|
"""
|
|
|
|
# 3. Send the request
|
|
headers = {"Content-Type": "application/xml"}
|
|
response = client.post("/api", content=xml_request, headers=headers)
|
|
|
|
# 4. Assert the response
|
|
assert response.status_code == 404
|
|
response_data = xmltodict.parse(response.content)
|
|
assert "Error" in response_data["AutoStore"]
|
|
assert "Bin not found" in response_data["AutoStore"]["Error"]["Description"]
|
|
|
|
|
|
def test_create_task_group_fails_if_port_not_found_xml(client, db_session):
|
|
"""Tests that creating a task group via XML fails if the port does not exist."""
|
|
# 1. Create Bins
|
|
crud.create_bin(db_session, schemas.BinCreate(id="BIN-TG-F-01", content_code=1))
|
|
db_session.commit()
|
|
|
|
# 2. Construct the XML request body with a non-existent port
|
|
xml_request = """
|
|
<AutoStore>
|
|
<CreateTaskgroup>
|
|
<Taskgroup>TG-FAIL-XML-01</Taskgroup>
|
|
<Port>NON-EXISTENT-PORT</Port>
|
|
<Tasks>
|
|
<Task>
|
|
<Task>T-FAIL-XML-01</Task>
|
|
<Bin>BIN-TG-F-01</Bin>
|
|
</Task>
|
|
</Tasks>
|
|
</CreateTaskgroup>
|
|
</AutoStore>
|
|
"""
|
|
|
|
# 3. Send the request
|
|
headers = {"Content-Type": "application/xml"}
|
|
response = client.post("/api", content=xml_request, headers=headers)
|
|
|
|
# 4. Assert the response
|
|
assert response.status_code == 404
|
|
response_data = xmltodict.parse(response.content)
|
|
assert "Error" in response_data["AutoStore"]
|
|
assert "Port not found" in response_data["AutoStore"]["Error"]["Description"]
|
|
|
|
|
|
def test_create_task_group_fails_if_bin_not_found_xml(client, db_session):
|
|
"""Tests that creating a task group via XML fails if a bin does not exist."""
|
|
# 1. Create Port
|
|
crud.create_port(db_session, schemas.PortCreate(id="P-TG-F-01", port_type="OUTBOUND"))
|
|
db_session.commit()
|
|
|
|
# 2. Construct the XML request body with a non-existent bin
|
|
xml_request = """
|
|
<AutoStore>
|
|
<CreateTaskgroup>
|
|
<Taskgroup>TG-FAIL-XML-02</Taskgroup>
|
|
<Port>P-TG-F-01</Port>
|
|
<Tasks>
|
|
<Task>
|
|
<Task>T-FAIL-XML-02</Task>
|
|
<Bin>NON-EXISTENT-BIN</Bin>
|
|
</Task>
|
|
</Tasks>
|
|
</CreateTaskgroup>
|
|
</AutoStore>
|
|
"""
|
|
|
|
# 3. Send the request
|
|
headers = {"Content-Type": "application/xml"}
|
|
response = client.post("/api", content=xml_request, headers=headers)
|
|
|
|
# 4. Assert the response
|
|
assert response.status_code == 404
|
|
response_data = xmltodict.parse(response.content)
|
|
assert "Error" in response_data["AutoStore"]
|
|
assert "Bin not found" in response_data["AutoStore"]["Error"]["Description"]
|
|
|
|
|
|
def test_create_taskgroup_xml(client, db_session):
|
|
"""Tests creating a task group via the XML interface."""
|
|
# 1. Create necessary Bins and Port in the database first
|
|
port_id = "P-XML-01"
|
|
bin_id_1 = "BIN-XML-01"
|
|
bin_id_2 = "BIN-XML-02"
|
|
crud.create_port(db_session, schemas.PortCreate(id=port_id, port_type="OUTBOUND"))
|
|
crud.create_bin(db_session, schemas.BinCreate(id=bin_id_1, content_code=1))
|
|
crud.create_bin(db_session, schemas.BinCreate(id=bin_id_2, content_code=1))
|
|
db_session.commit()
|
|
|
|
# 2. Construct the XML request body
|
|
xml_request = f"""
|
|
<AutoStore>
|
|
<CreateTaskgroup>
|
|
<Taskgroup>TG-XML-01</Taskgroup>
|
|
<Port>{port_id}</Port>
|
|
<Tasks>
|
|
<Task>
|
|
<Task>TASK-XML-01</Task>
|
|
<Bin>{bin_id_1}</Bin>
|
|
</Task>
|
|
<Task>
|
|
<Task>TASK-XML-02</Task>
|
|
<Bin>{bin_id_2}</Bin>
|
|
</Task>
|
|
</Tasks>
|
|
</CreateTaskgroup>
|
|
</AutoStore>
|
|
"""
|
|
|
|
# 3. Send the request to the /api endpoint
|
|
headers = {"Content-Type": "application/xml"}
|
|
response = client.post("/api", content=xml_request, headers=headers)
|
|
|
|
# 4. Assert the response
|
|
assert response.status_code == 200
|
|
assert response.headers["content-type"] == "application/xml"
|
|
|
|
# 5. Parse the XML response and validate its content
|
|
import xmltodict
|
|
response_data = xmltodict.parse(response.content)
|
|
assert "AutoStore" in response_data
|
|
assert "CreateTaskgroup" in response_data["AutoStore"]
|
|
assert "ExecutionTime" in response_data["AutoStore"]["CreateTaskgroup"]
|
|
|
|
# 6. Verify that the data was actually created in the database
|
|
db_task_group = crud.get_task_group(db_session, "TG-XML-01")
|
|
assert db_task_group is not None
|
|
assert db_task_group.id == "TG-XML-01"
|
|
assert db_task_group.port_id == port_id
|
|
assert len(db_task_group.tasks) == 2
|
|
assert db_task_group.tasks[0].id == "TASK-XML-01"
|
|
assert db_task_group.tasks[1].bin_id == bin_id_2
|
|
|
|
|
|
def test_get_task_info_for_group(client, db_session):
|
|
"""Tests the GetTaskInfo command for a whole task group."""
|
|
# 1. Create a task group first
|
|
port_id = "P-INFO-01"
|
|
bin_id = "BIN-INFO-01"
|
|
task_group_id = "TG-INFO-01"
|
|
task_id = "TASK-INFO-01"
|
|
crud.create_port(db_session, schemas.PortCreate(id=port_id, port_type="OUTBOUND"))
|
|
crud.create_bin(db_session, schemas.BinCreate(id=bin_id, content_code=1))
|
|
crud.create_task_group(
|
|
db_session,
|
|
schemas.TaskGroupCreate(
|
|
id=task_group_id,
|
|
port_id=port_id,
|
|
tasks=[schemas.TaskCreate(id=task_id, bin_id=bin_id, task_type="OUTBOUND")],
|
|
),
|
|
)
|
|
db_session.commit()
|
|
|
|
# 2. Construct the XML request
|
|
xml_request = f"""
|
|
<AutoStore>
|
|
<GetTaskInfo>
|
|
<Taskgroup>{task_group_id}</Taskgroup>
|
|
</GetTaskInfo>
|
|
</AutoStore>
|
|
"""
|
|
|
|
# 3. Send the request
|
|
headers = {"Content-Type": "application/xml"}
|
|
response = client.post("/api", content=xml_request, headers=headers)
|
|
|
|
# 4. Assert the response
|
|
assert response.status_code == 200
|
|
response_data = xmltodict.parse(response.content)
|
|
info = response_data["AutoStore"]["GetTaskInfo"]["Taskgroup"]
|
|
assert info["Taskgroup"] == task_group_id
|
|
assert info["Port"] == port_id
|
|
assert info["Status"] == "queued"
|
|
assert info["Tasks"]["Task"]["Task"] == task_id
|
|
|
|
|
|
def test_get_task_info_for_single_task(client, db_session):
|
|
"""Tests the GetTaskInfo command for a single task."""
|
|
# 1. Create a task group first
|
|
port_id = "P-INFO-02"
|
|
bin_id = "BIN-INFO-02"
|
|
task_group_id = "TG-INFO-02"
|
|
task_id = "TASK-INFO-02"
|
|
crud.create_port(db_session, schemas.PortCreate(id=port_id, port_type="OUTBOUND"))
|
|
crud.create_bin(db_session, schemas.BinCreate(id=bin_id, content_code=1))
|
|
crud.create_task_group(
|
|
db_session,
|
|
schemas.TaskGroupCreate(
|
|
id=task_group_id,
|
|
port_id=port_id,
|
|
tasks=[schemas.TaskCreate(id=task_id, bin_id=bin_id, task_type="OUTBOUND")],
|
|
),
|
|
)
|
|
db_session.commit()
|
|
|
|
# 2. Construct the XML request
|
|
xml_request = f"""
|
|
<AutoStore>
|
|
<GetTaskInfo>
|
|
<Task>{task_id}</Task>
|
|
</GetTaskInfo>
|
|
</AutoStore>
|
|
"""
|
|
|
|
# 3. Send the request
|
|
headers = {"Content-Type": "application/xml"}
|
|
response = client.post("/api", content=xml_request, headers=headers)
|
|
|
|
# 4. Assert the response
|
|
assert response.status_code == 200
|
|
response_data = xmltodict.parse(response.content)
|
|
info = response_data["AutoStore"]["GetTaskInfo"]["Task"]
|
|
assert info["Task"] == task_id
|
|
assert info["Bin"] == bin_id
|
|
assert info["Status"] == "queued"
|
|
assert info["Taskgroup"] == task_group_id
|
|
|
|
|
|
def test_get_task_info_not_found(client):
|
|
"""Tests that GetTaskInfo returns an error for a non-existent ID."""
|
|
xml_request = """
|
|
<AutoStore>
|
|
<GetTaskInfo>
|
|
<Taskgroup>NON-EXISTENT-GROUP</Taskgroup>
|
|
</GetTaskInfo>
|
|
</AutoStore>
|
|
"""
|
|
headers = {"Content-Type": "application/xml"}
|
|
response = client.post("/api", content=xml_request, headers=headers)
|
|
assert response.status_code == 404
|
|
response_data = xmltodict.parse(response.content)
|
|
assert "Taskgroup not found" in response_data["AutoStore"]["Error"]["Description"]
|
|
|
|
|
|
def test_get_bin_info_xml(client, db_session):
|
|
"""Tests the GetBinInfo command for a specific bin."""
|
|
# Arrange: Create a bin in the database
|
|
bin_id = "BIN-INFO-01"
|
|
bin_data = schemas.BinCreate(id=bin_id, content_code=5)
|
|
crud.create_bin(db_session, bin_data)
|
|
|
|
# Act: Send GetBinInfo XML request
|
|
xml_request = f"""
|
|
<AutoStore>
|
|
<GetBinInfo>
|
|
<Bin>{bin_id}</Bin>
|
|
</GetBinInfo>
|
|
</AutoStore>
|
|
"""
|
|
headers = {"Content-Type": "application/xml"}
|
|
response = client.post("/api", content=xml_request, headers=headers)
|
|
|
|
# Assert: Check response
|
|
assert response.status_code == 200
|
|
response_data = xmltodict.parse(response.content)
|
|
bin_info = response_data["AutoStore"]["GetBinInfo"]
|
|
assert bin_info["Bin"] == bin_id
|
|
assert bin_info["ContentCode"] == "5"
|
|
|
|
|
|
def test_get_bin_info_not_found_xml(client):
|
|
"""Tests that GetBinInfo returns an error for a non-existent bin."""
|
|
xml_request = """
|
|
<AutoStore>
|
|
<GetBinInfo>
|
|
<Bin>NON-EXISTENT-BIN</Bin>
|
|
</GetBinInfo>
|
|
</AutoStore>
|
|
"""
|
|
headers = {"Content-Type": "application/xml"}
|
|
response = client.post("/api", content=xml_request, headers=headers)
|
|
assert response.status_code == 404
|
|
response_data = xmltodict.parse(response.content)
|
|
assert "Bin not found" in response_data["AutoStore"]["Error"]["Description"]
|
|
|
|
|
|
def test_get_bin_info_missing_bin_id_xml(client):
|
|
"""Tests that GetBinInfo returns an error when Bin ID is missing."""
|
|
xml_request = """
|
|
<AutoStore>
|
|
<GetBinInfo>
|
|
</GetBinInfo>
|
|
</AutoStore>
|
|
"""
|
|
headers = {"Content-Type": "application/xml"}
|
|
response = client.post("/api", content=xml_request, headers=headers)
|
|
assert response.status_code == 400
|
|
response_data = xmltodict.parse(response.content)
|
|
assert "Bin ID is required" in response_data["AutoStore"]["Error"]["Description"]
|
|
|
|
|
|
def test_delete_taskgroup_xml(client, db_session):
|
|
"""Tests the DeleteTaskgroup command for an existing task group."""
|
|
# Arrange: Create a port, bin, and task group with tasks
|
|
port_id = "P-DELETE-01"
|
|
bin_id = "BIN-DELETE-01"
|
|
task_group_id = "TG-DELETE-01"
|
|
task_id = "TASK-DELETE-01"
|
|
|
|
port_data = schemas.PortCreate(id=port_id, port_type="OUTBOUND")
|
|
crud.create_port(db_session, port_data)
|
|
|
|
bin_data = schemas.BinCreate(id=bin_id, content_code=1)
|
|
crud.create_bin(db_session, bin_data)
|
|
|
|
task_group_data = schemas.TaskGroupCreate(
|
|
id=task_group_id,
|
|
port_id=port_id,
|
|
tasks=[schemas.TaskCreate(id=task_id, bin_id=bin_id, task_type="OUTBOUND")]
|
|
)
|
|
crud.create_task_group(db_session, task_group_data)
|
|
|
|
# Verify task group exists
|
|
task_group = crud.get_task_group(db_session, task_group_id)
|
|
assert task_group is not None
|
|
assert len(task_group.tasks) == 1
|
|
|
|
# Act: Send DeleteTaskgroup XML request
|
|
xml_request = f"""
|
|
<AutoStore>
|
|
<DeleteTaskgroup>
|
|
<Taskgroup>{task_group_id}</Taskgroup>
|
|
</DeleteTaskgroup>
|
|
</AutoStore>
|
|
"""
|
|
headers = {"Content-Type": "application/xml"}
|
|
response = client.post("/api", content=xml_request, headers=headers)
|
|
|
|
# Assert: Check response and verify deletion
|
|
assert response.status_code == 200
|
|
response_data = xmltodict.parse(response.content)
|
|
assert "DeleteTaskgroup" in response_data["AutoStore"]
|
|
assert response_data["AutoStore"]["DeleteTaskgroup"]["Success"] == "true"
|
|
|
|
# Verify task group and tasks are deleted
|
|
deleted_task_group = crud.get_task_group(db_session, task_group_id)
|
|
assert deleted_task_group is None
|
|
|
|
deleted_task = crud.get_task(db_session, task_id)
|
|
assert deleted_task is None
|
|
|
|
|
|
def test_delete_taskgroup_not_found_xml(client):
|
|
"""Tests that DeleteTaskgroup returns an error for a non-existent task group."""
|
|
xml_request = """
|
|
<AutoStore>
|
|
<DeleteTaskgroup>
|
|
<Taskgroup>NON-EXISTENT-GROUP</Taskgroup>
|
|
</DeleteTaskgroup>
|
|
</AutoStore>
|
|
"""
|
|
headers = {"Content-Type": "application/xml"}
|
|
response = client.post("/api", content=xml_request, headers=headers)
|
|
assert response.status_code == 404
|
|
response_data = xmltodict.parse(response.content)
|
|
assert "Taskgroup not found" in response_data["AutoStore"]["Error"]["Description"]
|
|
|
|
|
|
def test_delete_taskgroup_missing_id_xml(client):
|
|
"""Tests that DeleteTaskgroup returns an error when Taskgroup ID is missing."""
|
|
xml_request = """
|
|
<AutoStore>
|
|
<DeleteTaskgroup>
|
|
</DeleteTaskgroup>
|
|
</AutoStore>
|
|
"""
|
|
headers = {"Content-Type": "application/xml"}
|
|
response = client.post("/api", content=xml_request, headers=headers)
|
|
assert response.status_code == 400
|
|
response_data = xmltodict.parse(response.content)
|
|
assert "Taskgroup ID is required" in response_data["AutoStore"]["Error"]["Description"]
|
|
|
|
|
|
def test_get_next_task_xml(client, db_session):
|
|
"""Tests the GetNextTask command for a port with available tasks."""
|
|
# Arrange: Create a port, bin, and task group with tasks
|
|
port_id = "P-NEXT-01"
|
|
bin_id = "BIN-NEXT-01"
|
|
task_group_id = "TG-NEXT-01"
|
|
task_id = "TASK-NEXT-01"
|
|
|
|
port_data = schemas.PortCreate(id=port_id, port_type="OUTBOUND")
|
|
crud.create_port(db_session, port_data)
|
|
|
|
bin_data = schemas.BinCreate(id=bin_id, content_code=1)
|
|
crud.create_bin(db_session, bin_data)
|
|
|
|
task_group_data = schemas.TaskGroupCreate(
|
|
id=task_group_id,
|
|
port_id=port_id,
|
|
tasks=[schemas.TaskCreate(id=task_id, bin_id=bin_id, task_type="OUTBOUND")]
|
|
)
|
|
crud.create_task_group(db_session, task_group_data)
|
|
|
|
# Act: Send GetNextTask XML request
|
|
xml_request = f"""
|
|
<AutoStore>
|
|
<GetNextTask>
|
|
<Port>{port_id}</Port>
|
|
</GetNextTask>
|
|
</AutoStore>
|
|
"""
|
|
headers = {"Content-Type": "application/xml"}
|
|
response = client.post("/api", content=xml_request, headers=headers)
|
|
|
|
# Assert: Check response contains next task
|
|
assert response.status_code == 200
|
|
response_data = xmltodict.parse(response.content)
|
|
assert "GetNextTask" in response_data["AutoStore"]
|
|
next_task = response_data["AutoStore"]["GetNextTask"]
|
|
assert next_task["Task"] == task_id
|
|
assert next_task["Bin"] == bin_id
|
|
assert next_task["Taskgroup"] == task_group_id
|
|
|
|
|
|
def test_get_next_task_no_tasks_available_xml(client, db_session):
|
|
"""Tests that GetNextTask returns no task when none are available for the port."""
|
|
# Arrange: Create a port but no tasks for it
|
|
port_id = "P-NEXT-02"
|
|
port_data = schemas.PortCreate(id=port_id, port_type="OUTBOUND")
|
|
crud.create_port(db_session, port_data)
|
|
|
|
# Act: Send GetNextTask XML request
|
|
xml_request = f"""
|
|
<AutoStore>
|
|
<GetNextTask>
|
|
<Port>{port_id}</Port>
|
|
</GetNextTask>
|
|
</AutoStore>
|
|
"""
|
|
headers = {"Content-Type": "application/xml"}
|
|
response = client.post("/api", content=xml_request, headers=headers)
|
|
|
|
# Assert: Check response indicates no task available
|
|
assert response.status_code == 200
|
|
response_data = xmltodict.parse(response.content)
|
|
assert "GetNextTask" in response_data["AutoStore"]
|
|
assert response_data["AutoStore"]["GetNextTask"]["NoTask"] == "true"
|
|
|
|
|
|
def test_get_next_task_port_not_found_xml(client):
|
|
"""Tests that GetNextTask returns an error for a non-existent port."""
|
|
xml_request = """
|
|
<AutoStore>
|
|
<GetNextTask>
|
|
<Port>NON-EXISTENT-PORT</Port>
|
|
</GetNextTask>
|
|
</AutoStore>
|
|
"""
|
|
headers = {"Content-Type": "application/xml"}
|
|
response = client.post("/api", content=xml_request, headers=headers)
|
|
assert response.status_code == 404
|
|
response_data = xmltodict.parse(response.content)
|
|
assert "Port not found" in response_data["AutoStore"]["Error"]["Description"]
|
|
|
|
|
|
def test_get_next_task_missing_port_xml(client):
|
|
"""Tests that GetNextTask returns an error when Port ID is missing."""
|
|
xml_request = """
|
|
<AutoStore>
|
|
<GetNextTask>
|
|
</GetNextTask>
|
|
</AutoStore>
|
|
"""
|
|
headers = {"Content-Type": "application/xml"}
|
|
response = client.post("/api", content=xml_request, headers=headers)
|
|
assert response.status_code == 400
|
|
response_data = xmltodict.parse(response.content)
|
|
assert "Port ID is required" in response_data["AutoStore"]["Error"]["Description"]
|
|
|
|
|
|
def test_set_port_mode_xml(client, db_session):
|
|
"""Tests the SetPortMode command to change a port's mode."""
|
|
# Arrange: Create a port
|
|
port_id = "P-MODE-01"
|
|
port_data = schemas.PortCreate(id=port_id, port_type="INBOUND")
|
|
crud.create_port(db_session, port_data)
|
|
|
|
# Verify initial port type
|
|
port = crud.get_port(db_session, port_id)
|
|
assert port.port_type == "INBOUND"
|
|
|
|
# Act: Send SetPortMode XML request to change to OUTBOUND
|
|
xml_request = f"""
|
|
<AutoStore>
|
|
<SetPortMode>
|
|
<Port>{port_id}</Port>
|
|
<Mode>OUTBOUND</Mode>
|
|
</SetPortMode>
|
|
</AutoStore>
|
|
"""
|
|
headers = {"Content-Type": "application/xml"}
|
|
response = client.post("/api", content=xml_request, headers=headers)
|
|
|
|
# Assert: Check response and verify port mode changed
|
|
assert response.status_code == 200
|
|
response_data = xmltodict.parse(response.content)
|
|
assert "SetPortMode" in response_data["AutoStore"]
|
|
assert response_data["AutoStore"]["SetPortMode"]["Success"] == "true"
|
|
|
|
# Verify port mode was actually changed in database
|
|
updated_port = crud.get_port(db_session, port_id)
|
|
assert updated_port.port_type == "OUTBOUND"
|
|
|
|
|
|
def test_set_port_mode_to_off_xml(client, db_session):
|
|
"""Tests the SetPortMode command to change a port's mode to OFF."""
|
|
# Arrange: Create a port
|
|
port_id = "P-MODE-02"
|
|
port_data = schemas.PortCreate(id=port_id, port_type="OUTBOUND")
|
|
crud.create_port(db_session, port_data)
|
|
|
|
# Act: Send SetPortMode XML request to change to OFF
|
|
xml_request = f"""
|
|
<AutoStore>
|
|
<SetPortMode>
|
|
<Port>{port_id}</Port>
|
|
<Mode>OFF</Mode>
|
|
</SetPortMode>
|
|
</AutoStore>
|
|
"""
|
|
headers = {"Content-Type": "application/xml"}
|
|
response = client.post("/api", content=xml_request, headers=headers)
|
|
|
|
# Assert: Check response and verify port mode changed
|
|
assert response.status_code == 200
|
|
response_data = xmltodict.parse(response.content)
|
|
assert "SetPortMode" in response_data["AutoStore"]
|
|
assert response_data["AutoStore"]["SetPortMode"]["Success"] == "true"
|
|
|
|
# Verify port mode was actually changed in database
|
|
updated_port = crud.get_port(db_session, port_id)
|
|
assert updated_port.port_type == "OFF"
|
|
|
|
|
|
def test_set_port_mode_invalid_mode_xml(client, db_session):
|
|
"""Tests that SetPortMode returns an error for an invalid mode."""
|
|
# Arrange: Create a port
|
|
port_id = "P-MODE-03"
|
|
port_data = schemas.PortCreate(id=port_id, port_type="INBOUND")
|
|
crud.create_port(db_session, port_data)
|
|
|
|
# Act: Send SetPortMode XML request with invalid mode
|
|
xml_request = f"""
|
|
<AutoStore>
|
|
<SetPortMode>
|
|
<Port>{port_id}</Port>
|
|
<Mode>INVALID_MODE</Mode>
|
|
</SetPortMode>
|
|
</AutoStore>
|
|
"""
|
|
headers = {"Content-Type": "application/xml"}
|
|
response = client.post("/api", content=xml_request, headers=headers)
|
|
|
|
# Assert: Check response contains error
|
|
assert response.status_code == 400
|
|
response_data = xmltodict.parse(response.content)
|
|
assert "Invalid port mode" in response_data["AutoStore"]["Error"]["Description"]
|
|
|
|
|
|
def test_set_port_mode_port_not_found_xml(client):
|
|
"""Tests that SetPortMode returns an error for a non-existent port."""
|
|
xml_request = """
|
|
<AutoStore>
|
|
<SetPortMode>
|
|
<Port>NON-EXISTENT-PORT</Port>
|
|
<Mode>OUTBOUND</Mode>
|
|
</SetPortMode>
|
|
</AutoStore>
|
|
"""
|
|
headers = {"Content-Type": "application/xml"}
|
|
response = client.post("/api", content=xml_request, headers=headers)
|
|
assert response.status_code == 404
|
|
response_data = xmltodict.parse(response.content)
|
|
assert "Port not found" in response_data["AutoStore"]["Error"]["Description"]
|
|
|
|
|
|
def test_set_port_mode_missing_port_xml(client):
|
|
"""Tests that SetPortMode returns an error when Port ID is missing."""
|
|
xml_request = """
|
|
<AutoStore>
|
|
<SetPortMode>
|
|
<Mode>OUTBOUND</Mode>
|
|
</SetPortMode>
|
|
</AutoStore>
|
|
"""
|
|
headers = {"Content-Type": "application/xml"}
|
|
response = client.post("/api", content=xml_request, headers=headers)
|
|
assert response.status_code == 400
|
|
response_data = xmltodict.parse(response.content)
|
|
assert "Port ID is required" in response_data["AutoStore"]["Error"]["Description"]
|
|
|
|
|
|
def test_set_port_mode_missing_mode_xml(client, db_session):
|
|
"""Tests that SetPortMode returns an error when Mode is missing."""
|
|
# Arrange: Create a port
|
|
port_id = "P-MODE-04"
|
|
port_data = schemas.PortCreate(id=port_id, port_type="INBOUND")
|
|
crud.create_port(db_session, port_data)
|
|
|
|
xml_request = f"""
|
|
<AutoStore>
|
|
<SetPortMode>
|
|
<Port>{port_id}</Port>
|
|
</SetPortMode>
|
|
</AutoStore>
|
|
"""
|
|
headers = {"Content-Type": "application/xml"}
|
|
response = client.post("/api", content=xml_request, headers=headers)
|
|
assert response.status_code == 400
|
|
response_data = xmltodict.parse(response.content)
|
|
assert "Mode is required" in response_data["AutoStore"]["Error"]["Description"]
|