Skip to content
This repository was archived by the owner on Feb 27, 2025. It is now read-only.

Commit fae6bdc

Browse files
ArtemConstantinovArtemConstantinov
committed
add tessts, expose up to 35 types via apply method
1 parent 8657baf commit fae6bdc

27 files changed

+1462
-188
lines changed

‎requirements.txt

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
typing_extensions>=4.3.0
22
loguru
33
pdoc
4-
pytest
4+
pytest
5+
pytest-asyncio

‎tests/test_hli.py

Lines changed: 95 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,95 @@
1+
from __future__ import annotations
2+
3+
import timeit
4+
import pstats
5+
import cProfile
6+
7+
import asyncio
8+
import os
9+
import sys
10+
import logging
11+
from typing import Coroutine, List
12+
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
13+
14+
from loguru import logger # noqa: E402
15+
from xoa_driver.testers import L23Tester # noqa: E402
16+
from xoa_driver.lli import establish_connection # noqa: E402
17+
from xoa_driver.lli import commands # noqa: E402
18+
from xoa_driver.lli import TransportationHandler # noqa: E402
19+
from xoa_driver.utils import apply, apply_iter # noqa: E402
20+
from xoa_driver.internals.core.transporter.protocol.payload import ResponseBodyStruct, field, XmpSequence, XmpByte
21+
# TODO: <GET> response contain more fields then driver can parse
22+
# TODO: <GET> response contains less fields then driver can parse
23+
# TODO: <SET> request require more fields then server expecting
24+
# TODO: <SET> request require less fields then server expects
25+
26+
27+
async def test_command() -> None:
28+
data = b'\x00\x02\x00\x06\x00\x06\x00\x06'
29+
30+
class GetDataAttr(ResponseBodyStruct):
31+
port_counts: List[int] = field(XmpSequence(types_chunk=[XmpByte()]))
32+
33+
obj = GetDataAttr(data)
34+
print(obj)
35+
36+
37+
async def test_lli() -> None:
38+
# logging.basicConfig(
39+
# format='%(relativeCreated)5d %(name)-15s %(levelname)-8s %(message)s',
40+
# level=logging.DEBUG
41+
# )
42+
# logger_ = logging.getLogger(__file__)
43+
ctx = TransportationHandler(enable_logging=True, custom_logger=logger)
44+
await establish_connection(ctx, "192.168.1.197")
45+
# # print("Is connected", ctx.is_connected)
46+
# with cProfile.Profile() as pr:
47+
*_, pc = await apply(
48+
commands.C_LOGON(ctx).set("xena"),
49+
commands.C_OWNER(ctx).set("xoa"),
50+
commands.C_OWNER(ctx).get(),
51+
# commands.M_CAPABILITIES(ctx, 1).get(),
52+
# commands.P_CAPABILITIES(ctx, 1, 1).get(),
53+
commands.C_PORTCOUNTS(ctx).get()
54+
)
55+
print(pc._order.field_names, pc.to_bytes())
56+
# req = apply_iter(*[commands.P_CAPABILITIES(ctx, 1, 1).get() for _ in range(1_000)])
57+
# async for resp in req:
58+
# resp.tx_eq_tap_max_val
59+
# stats = pstats.Stats(pr)
60+
# stats.sort_stats(pstats.SortKey.TIME)
61+
# stats.print_stats(20)
62+
ctx.close()
63+
64+
# port = PortL23()
65+
# if isinstance(port, PortL23) and port.is_capable_of(ANLT, PCS_PMA)
66+
67+
# Testsuite 2544:
68+
# required_port_type: L23Port
69+
# required_functionalities:
70+
# ANLT functionalities:
71+
# required_port_type: L23Port,
72+
# required_functionalities: ANLT
73+
74+
75+
async def test_hli() -> None:
76+
async with L23Tester("demo.xenanetworks.com", "ACO") as tester:
77+
t_cap = await tester.capabilities.get()
78+
print(t_cap.can_sync_traffic_start)
79+
80+
81+
def run(method: Coroutine) -> None:
82+
import platform
83+
if platform.system() == 'Windows':
84+
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
85+
asyncio.run(method)
86+
87+
88+
if __name__ == "__main__":
89+
run(test_hli())
90+
# result = timeit.timeit(
91+
# "run(main())",
92+
# setup="from __main__ import run, main",
93+
# number=10
94+
# )
95+
# print(result)

‎tests/test_lli.py

Lines changed: 95 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,95 @@
1+
from __future__ import annotations
2+
3+
import timeit
4+
import pstats
5+
import cProfile
6+
7+
import asyncio
8+
import os
9+
import sys
10+
import logging
11+
from typing import Coroutine, List
12+
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
13+
14+
from loguru import logger # noqa: E402
15+
from xoa_driver.testers import L23Tester # noqa: E402
16+
from xoa_driver.lli import establish_connection # noqa: E402
17+
from xoa_driver.lli import commands # noqa: E402
18+
from xoa_driver.lli import TransportationHandler # noqa: E402
19+
from xoa_driver.utils import apply, apply_iter # noqa: E402
20+
from xoa_driver.internals.core.transporter.protocol.payload import ResponseBodyStruct, field, XmpSequence, XmpByte
21+
# TODO: <GET> response contain more fields then driver can parse
22+
# TODO: <GET> response contains less fields then driver can parse
23+
# TODO: <SET> request require more fields then server expecting
24+
# TODO: <SET> request require less fields then server expects
25+
26+
27+
async def test_command() -> None:
28+
data = b'\x00\x02\x00\x06\x00\x06\x00\x06'
29+
30+
class GetDataAttr(ResponseBodyStruct):
31+
port_counts: List[int] = field(XmpSequence(types_chunk=[XmpByte()]))
32+
33+
obj = GetDataAttr(data)
34+
print(obj)
35+
36+
37+
async def test_lli() -> None:
38+
# logging.basicConfig(
39+
# format='%(relativeCreated)5d %(name)-15s %(levelname)-8s %(message)s',
40+
# level=logging.DEBUG
41+
# )
42+
# logger_ = logging.getLogger(__file__)
43+
ctx = TransportationHandler(enable_logging=True, custom_logger=logger)
44+
await establish_connection(ctx, "192.168.1.197")
45+
# # print("Is connected", ctx.is_connected)
46+
# with cProfile.Profile() as pr:
47+
*_, pc = await apply(
48+
commands.C_LOGON(ctx).set("xena"),
49+
commands.C_OWNER(ctx).set("xoa"),
50+
commands.C_OWNER(ctx).get(),
51+
# commands.M_CAPABILITIES(ctx, 1).get(),
52+
# commands.P_CAPABILITIES(ctx, 1, 1).get(),
53+
commands.C_PORTCOUNTS(ctx).get()
54+
)
55+
print(pc._order.field_names, pc.to_bytes())
56+
# req = apply_iter(*[commands.P_CAPABILITIES(ctx, 1, 1).get() for _ in range(1_000)])
57+
# async for resp in req:
58+
# resp.tx_eq_tap_max_val
59+
# stats = pstats.Stats(pr)
60+
# stats.sort_stats(pstats.SortKey.TIME)
61+
# stats.print_stats(20)
62+
ctx.close()
63+
64+
# port = PortL23()
65+
# if isinstance(port, PortL23) and port.is_capable_of(ANLT, PCS_PMA)
66+
67+
# Testsuite 2544:
68+
# required_port_type: L23Port
69+
# required_functionalities:
70+
# ANLT functionalities:
71+
# required_port_type: L23Port,
72+
# required_functionalities: ANLT
73+
74+
75+
async def test_hli() -> None:
76+
async with L23Tester("demo.xenanetworks.com", "ACO") as tester:
77+
t_cap = await tester.capabilities.get()
78+
print(t_cap.can_sync_traffic_start)
79+
80+
81+
def run(method: Coroutine) -> None:
82+
import platform
83+
if platform.system() == 'Windows':
84+
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
85+
asyncio.run(method)
86+
87+
88+
if __name__ == "__main__":
89+
run(test_hli())
90+
# result = timeit.timeit(
91+
# "run(main())",
92+
# setup="from __main__ import run, main",
93+
# number=10
94+
# )
95+
# print(result)

0 commit comments

Comments
 (0)