107 lines
3.6 KiB
Python
107 lines
3.6 KiB
Python
from app.models.attachment_point_name import AttachmentPointName
|
|
from app.models.augmented_part import AugmentedPart
|
|
from app.models.iso_thread import IsoThread
|
|
from app.models.manufacturer import Manufacturer
|
|
from app.models.part_colour import PartColour
|
|
from app.models.part_type import PartType
|
|
|
|
|
|
async def _seed_part(db):
|
|
manufacturer = Manufacturer(name="Acme", url="https://acme.test")
|
|
colour = PartColour(name="Black")
|
|
part_type = PartType(name="Flip Cover")
|
|
apn = AttachmentPointName(name="Mount")
|
|
iso = IsoThread(diameter_mm=52.0, pitch_mm=0.75)
|
|
db.add_all([manufacturer, colour, part_type, apn, iso])
|
|
await db.flush()
|
|
|
|
part = AugmentedPart(
|
|
description="Flip cover",
|
|
colour_id=colour.id,
|
|
manufacturer_id=manufacturer.id,
|
|
part_type_id=part_type.id,
|
|
)
|
|
db.add(part)
|
|
await db.flush()
|
|
await db.commit()
|
|
return part.id, apn.id, iso.id
|
|
|
|
|
|
async def _token(client, email, password="secret123"):
|
|
resp = await client.post(
|
|
"/api/auth/login", data={"username": email, "password": password}
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
return resp.json()["access_token"]
|
|
|
|
|
|
async def test_create_manual_attachment_point_with_no_connection_succeeds(
|
|
client, seeded, make_user, db_sessionmaker
|
|
):
|
|
await make_user("root@example.com", is_superuser=True)
|
|
headers = {"Authorization": f"Bearer {await _token(client, 'root@example.com')}"}
|
|
async with db_sessionmaker() as db:
|
|
part_id, apn_id, _ = await _seed_part(db)
|
|
|
|
resp = await client.post(
|
|
"/api/attachment-points",
|
|
json={
|
|
"augmented_part_id": part_id,
|
|
"attachment_point_name_id": apn_id,
|
|
},
|
|
headers=headers,
|
|
)
|
|
assert resp.status_code == 201, resp.text
|
|
body = resp.json()
|
|
assert body["iso_thread_id"] is None
|
|
assert body["unc_thread_id"] is None
|
|
assert body["clamp_connection_id"] is None
|
|
assert body["static_connection_id"] is None
|
|
|
|
|
|
async def test_create_attachment_point_with_two_connections_rejected(
|
|
client, seeded, make_user, db_sessionmaker
|
|
):
|
|
await make_user("root2@example.com", is_superuser=True)
|
|
headers = {"Authorization": f"Bearer {await _token(client, 'root2@example.com')}"}
|
|
async with db_sessionmaker() as db:
|
|
part_id, apn_id, iso_id = await _seed_part(db)
|
|
from app.crud import static_connection as static_connection_crud
|
|
|
|
static = await static_connection_crud.create(db, od_mm=40.0, od_in=1.57)
|
|
static_id = static.id
|
|
|
|
resp = await client.post(
|
|
"/api/attachment-points",
|
|
json={
|
|
"augmented_part_id": part_id,
|
|
"attachment_point_name_id": apn_id,
|
|
"iso_thread_id": iso_id,
|
|
"static_connection_id": static_id,
|
|
},
|
|
headers=headers,
|
|
)
|
|
assert resp.status_code == 400
|
|
assert "at most one" in resp.json()["detail"].lower()
|
|
|
|
|
|
async def test_create_attachment_point_with_one_connection_still_works(
|
|
client, seeded, make_user, db_sessionmaker
|
|
):
|
|
await make_user("root3@example.com", is_superuser=True)
|
|
headers = {"Authorization": f"Bearer {await _token(client, 'root3@example.com')}"}
|
|
async with db_sessionmaker() as db:
|
|
part_id, apn_id, iso_id = await _seed_part(db)
|
|
|
|
resp = await client.post(
|
|
"/api/attachment-points",
|
|
json={
|
|
"augmented_part_id": part_id,
|
|
"attachment_point_name_id": apn_id,
|
|
"iso_thread_id": iso_id,
|
|
},
|
|
headers=headers,
|
|
)
|
|
assert resp.status_code == 201, resp.text
|
|
assert resp.json()["iso_thread_id"] == iso_id
|