Skip to content
Projects
Groups
Snippets
Help
Loading...
Help
Support
Keyboard shortcuts
?
Submit feedback
Contribute to GitLab
Sign in / Register
Toggle navigation
slapos
Project overview
Project overview
Details
Activity
Releases
Repository
Repository
Files
Commits
Branches
Tags
Contributors
Graph
Compare
Issues
0
Issues
0
List
Boards
Labels
Milestones
Merge Requests
0
Merge Requests
0
CI / CD
CI / CD
Pipelines
Jobs
Schedules
Analytics
Analytics
CI / CD
Repository
Value Stream
Wiki
Wiki
Snippets
Snippets
Members
Members
Collapse sidebar
Close sidebar
Activity
Graph
Create a new issue
Jobs
Commits
Issue Boards
Open sidebar
Lu Xu
slapos
Commits
fcd95550
Commit
fcd95550
authored
Apr 11, 2024
by
Lu Xu
👀
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
fixup
parent
31983379
Changes
1
Show whitespace changes
Inline
Side-by-side
Showing
1 changed file
with
29 additions
and
26 deletions
+29
-26
software/end-to-end-testing/tests/test_new_lopcomm.py
software/end-to-end-testing/tests/test_new_lopcomm.py
+29
-26
No files found.
software/end-to-end-testing/tests/test_new_lopcomm.py
View file @
fcd95550
...
...
@@ -9,19 +9,13 @@ class WebsocketTestClass(e2e.EndToEndTestCase):
try
:
super
().
setUpClass
()
cls
.
enb_instance_name
=
"e2e-cb003-enb-kirr"
cls
.
ru1_instance_name
=
"e2e-cb003-ru1-kirr"
cls
.
cell1_instance_name
=
"e2e-cb003-cell1-kirr"
cls
.
cn_instance_name
=
"e2e-cb003-core-network-kirr"
cls
.
sim_instance_name
=
"e2e-cb003-sim-kirr"
cls
.
ue_instance_name
=
"e2e-sb005-ue-kirr"
cls
.
ue_cell1_instance_name
=
"e2e-sb005-ue-cell1-kirr"
cls
.
ru_instance_name
=
"e2e-cb003-ru-kirr"
# cls.enb_instance_name = time.strftime('e2e-cb003-enb-%Y-%B-%d-%H:%M:%S')
# cls.cn_instance_name = time.strftime('e2e-cb003-core-network-%Y-%B-%d-%H:%M:%S')
# cls.sim_instance_name = time.strftime('e2e-cb003-sim-%Y-%B-%d-%H:%M:%S')
# cls.ue_instance_name = time.strftime('e2e-sb005-ue-%Y-%B-%d-%H:%M:%S')
cls
.
enb_instance_name
=
time
.
strftime
(
'e2e-cb003-enb-new-%Y-%B-%d-%H:%M:%S'
)
cls
.
ru1_instance_name
=
time
.
strftime
(
'e2e-cb003-ru1-new-%Y-%B-%d-%H:%M:%S'
)
cls
.
cell1_instance_name
=
time
.
strftime
(
'e2e-cb003-cell1-new-%Y-%B-%d-%H:%M:%S'
)
cls
.
cn_instance_name
=
time
.
strftime
(
'e2e-cb003-core-network-new-%Y-%B-%d-%H:%M:%S'
)
cls
.
sim_instance_name
=
time
.
strftime
(
'e2e-cb003-sim-new-%Y-%B-%d-%H:%M:%S'
)
cls
.
ue_instance_name
=
time
.
strftime
(
'e2e-sb005-ue-new-%Y-%B-%d-%H:%M:%S'
)
cls
.
ue_cell1_instance_name
=
time
.
strftime
(
'e2e-sb005-ue-cell1-new-%Y-%B-%d-%H:%M:%S'
)
cls
.
product
=
"https://lab.nexedi.com/nexedi/slapos/-/raw/master/software/ors-amarisoft/software.cfg"
...
...
@@ -38,9 +32,10 @@ class WebsocketTestClass(e2e.EndToEndTestCase):
# Setup instances
cls
.
setup_instances
()
#
cls.waitUntilGreen(cls.enb_instance_name)
#cls.waitUntilGreen(cls.enb_instance_name)
cls
.
waitUntilGreen
(
cls
.
cn_instance_name
)
except
Exception
as
e
:
cls
.
logger
.
error
(
"Error during setup: "
+
str
(
e
))
# Ensure cleanup
...
...
@@ -61,16 +56,27 @@ class WebsocketTestClass(e2e.EndToEndTestCase):
@
classmethod
def
setup_instances
(
cls
):
cls
.
request_enb
()
cls
.
request_core_network
()
cls
.
wait_for_attribute
(
'cn_ipv6'
)
cls
.
request_enb
()
cls
.
request_ue
()
cls
.
setup_websocket_connection
()
@
classmethod
def
wait_for_attribute
(
cls
,
attr_name
,
timeout
=
600
):
start_time
=
time
.
time
()
while
not
hasattr
(
cls
,
attr_name
)
or
getattr
(
cls
,
attr_name
)
is
None
:
if
time
.
time
()
-
start_time
>
timeout
:
raise
TimeoutError
(
f"Timeout waiting for attribute '
{
attr_name
}
' to become available."
)
time
.
sleep
(
1
)
cls
.
logger
.
info
(
f"Attribute '
{
attr_name
}
' is now available."
)
@
classmethod
def
request_enb
(
cls
,
custom_params
=
None
):
cls
.
logger
.
info
(
"Request "
+
cls
.
enb_instance_name
)
enb_parameters
=
{
"enb_id"
:
"0x1A2D0"
,
"mme_list"
:
{
'1'
:
{
'mme_addr'
:
"[2a11:9ac0:11::10]"
}},
"mme_list"
:
{
'1'
:
{
'mme_addr'
:
cls
.
cn_ipv6
}},
"plmn_list"
:
{
"Australia"
:
{
"plmn"
:
"50501"
}}
}
...
...
@@ -170,6 +176,7 @@ class WebsocketTestClass(e2e.EndToEndTestCase):
if
core_network_instance
:
instance_infos
=
cls
.
getInstanceInfos
(
cls
.
cn_instance_name
)
cls
.
cn_instance_guid
=
instance_infos
.
news
[
'instance'
][
0
][
'reference'
]
cls
.
cn_ipv6
=
instance_infos
.
connection_dict
.
get
(
'core-network-ipv6'
)
cls
.
request_demo_sim_cards
()
return
True
return
False
...
...
@@ -198,13 +205,8 @@ class WebsocketTestClass(e2e.EndToEndTestCase):
@
classmethod
def
setup_websocket_connection
(
cls
):
ue_instance
=
cls
.
retry_request
(
cls
.
request_ue
)
cls
.
waitUntilGreen
(
cls
.
ue_instance_name
)
cls
.
ue_com_addr
=
ue_instance
.
get
(
'com_addr'
)
if
ue_instance
else
None
if
not
cls
.
ue_com_addr
:
cls
.
logger
.
error
(
"Failed to obtain UE com address."
)
return
#cls.waitUntilGreen(cls.ue_instance_name)
cls
.
wait_for_attribute
(
'ue_com_addr'
)
cls
.
ws_url
=
f"ws://
{
cls
.
ue_com_addr
}
"
cls
.
logger
.
info
(
f"Websocket URL:
{
cls
.
ws_url
}
"
)
...
...
@@ -237,10 +239,11 @@ class WebsocketTestClass(e2e.EndToEndTestCase):
ue_instance
=
cls
.
request
(
cls
.
product
,
cls
.
ue_instance_name
,
filter_kw
=
{
"computer_guid"
:
cls
.
comp_ue
},
partition_parameter_kw
=
{
'_'
:
ue_parameters
},
software_type
=
'ue'
)
software_type
=
'ue'
,
state
=
'started'
)
if
ue_instance
:
instance_infos
=
cls
.
getInstanceInfos
(
cls
.
ue_instance_name
)
cls
.
ue_instance_guid
=
instance_infos
.
news
[
'instance'
][
0
][
'reference'
]
cls
.
ue_com_addr
=
instance_infos
.
connection_dict
.
get
(
'com_addr'
)
cls
.
request_ue_cell1
()
return
True
return
False
...
...
@@ -257,7 +260,7 @@ class WebsocketTestClass(e2e.EndToEndTestCase):
'ru'
:
{
'ru_type'
:
'sdr'
,
'ru_link_type'
:
'sdr'
,
'sdr_dev_list'
:
[
1
],
'sdr_dev_list'
:
[
0
],
'n_antenna_dl'
:
1
,
'n_antenna_ul'
:
1
,
'tx_gain'
:
60
,
...
...
@@ -329,7 +332,7 @@ class BBUTest(WebsocketTestClass):
def
test_max_rx_sample_db
(
self
):
custom_params
=
{
"max_rx_sample_db"
:
-
99
}
BBUTest
.
request_enb
(
custom_params
)
self
.
waitUntilPromises
(
BBUTest
.
enb_instance_name
,
promise_name
=
"check
-rx-saturated"
,
expected
=
False
)
self
.
waitUntilPromises
(
BBUTest
.
enb_instance_name
,
promise_name
=
BBUTest
.
ru1_instance_name
+
"
-rx-saturated"
,
expected
=
False
)
def
test_min_rxtx_delay
(
self
):
custom_params
=
{
"min_rxtx_delay"
:
99
}
...
...
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment