Commit 21ecc7b7 authored by Lluis Gifre Renom's avatar Lluis Gifre Renom
Browse files

Context component: 

- Fixed start points of codec for spectrum occupancy used in OpticalLinkModel
- Updated unitary test
parent ad701351

src/context/service/database/models/Slot.py

+3 −3
Original line number Diff line number Diff line
@@ -68,15 +68,15 @@ class SlotType(TypeDecorator): 




class C_Slot(SlotType):

    start_point = 0

    start_point = 1

    width = 320





class L_Slot(SlotType):

    start_point = 0

    start_point = 101

    width = 550





class S_Slot(SlotType):

    start_point = 0

    start_point = 501

    width = 720

src/context/tests/test_optical_link_slots.py

+18 −16
Original line number Diff line number Diff line
@@ -34,9 +34,9 @@ class SlotSmokeModel(Base): 
    s_slots = Column(S_Slot, nullable=True)





def build_expected_slot_map(width: int, active_slots):

def build_expected_slot_map(start_slot: int, width: int, active_slots):

    active_slots = set(active_slots)

    return {str(slot): (1 if slot in active_slots else 0) for slot in range(width)}

    return {str(slot): (1 if slot in active_slots else 0) for slot in range(start_slot, start_slot + width)}





def build_sparse_slot_input(active_slots):
@@ -46,9 +46,9 @@ def build_sparse_slot_input(active_slots): 
@pytest.mark.parametrize(

    'slot_type,width,active_slots',

    [

        (C_Slot(), 320, [0, 17, 319]),

        (L_Slot(), 550, [0, 101, 549]),

        (S_Slot(), 720, [0, 205, 719]),

        (C_Slot(), 320, [1, 18, 320]),

        (L_Slot(), 550, [101, 202, 650]),

        (S_Slot(), 720, [501, 706, 1220]),

    ],

)

def test_slot_type_roundtrip_preserves_positions(slot_type, width, active_slots) -> None:
@@ -57,15 +57,15 @@ def test_slot_type_roundtrip_preserves_positions(slot_type, width, active_slots) 
    decoded = slot_type.process_result_value(encoded, dialect=None)



    assert encoded is not None

    assert decoded == build_expected_slot_map(width, active_slots)

    assert decoded == build_expected_slot_map(slot_type.start_point, width, active_slots)





@pytest.mark.parametrize(

    'slot_type,invalid_key',

    [

        (C_Slot(), 320),

        (L_Slot(), 550),

        (S_Slot(), 720),

        (C_Slot(), 321),

        (L_Slot(), 651),

        (S_Slot(), 1221),

    ],

)

def test_slot_type_rejects_out_of_range_keys(slot_type, invalid_key: int) -> None:
@@ -76,9 +76,9 @@ def test_slot_type_rejects_out_of_range_keys(slot_type, invalid_key: int) -> Non 
def _run_slot_smoke_test(engine: sqlalchemy.engine.Engine) -> None:

    Base.metadata.create_all(engine)

    try:

        c_slots = build_sparse_slot_input([0, 10, 319])

        l_slots = build_sparse_slot_input([0, 12, 549])

        s_slots = build_sparse_slot_input([0, 14, 719])

        c_slots = build_sparse_slot_input([1, 11, 320])

        l_slots = build_sparse_slot_input([101, 113, 650])

        s_slots = build_sparse_slot_input([501, 515, 1220])



        with Session(engine) as session:

            session.add(SlotSmokeModel(id=1, c_slots=c_slots, l_slots=l_slots, s_slots=s_slots))
@@ -86,9 +86,9 @@ def _run_slot_smoke_test(engine: sqlalchemy.engine.Engine) -> None: 


        with Session(engine) as session:

            stored = session.query(SlotSmokeModel).filter_by(id=1).one()

            assert stored.c_slots == build_expected_slot_map(320, [0, 10, 319])

            assert stored.l_slots == build_expected_slot_map(550, [0, 12, 549])

            assert stored.s_slots == build_expected_slot_map(720, [0, 14, 719])

            assert stored.c_slots == build_expected_slot_map(1, 320, [1, 11, 320])

            assert stored.l_slots == build_expected_slot_map(101, 550, [101, 113, 650])

            assert stored.s_slots == build_expected_slot_map(501, 720, [501, 515, 1220])

    finally:

        Base.metadata.drop_all(engine)
@@ -99,7 +99,9 @@ def test_slot_smoke_sqlite() -> None: 




def test_slot_smoke_cockroachdb() -> None:

    crdb_uri = os.environ['CRDB_URI']

    crdb_uri = os.environ.get('CRDB_URI')

    if crdb_uri is None:

        pytest.skip('CRDB_URI is not set')

    engine = sqlalchemy.create_engine(

        crdb_uri, connect_args={'application_name': 'tfs-slot-smoketest'}, future=True

    )