@@ -46,81 +46,80 @@ def await_nrfcloud(func, expected, field, timeout):
46
46
if expected in data :
47
47
break
48
48
49
- def get_appversion (t91x_fota ):
50
- shadow = t91x_fota .fota .get_device (t91x_fota .device_id )
49
+ def get_appversion (dut_fota ):
50
+ shadow = dut_fota .fota .get_device (dut_fota .device_id )
51
51
return shadow ["state" ]["reported" ]["device" ]["deviceInfo" ]["appVersion" ]
52
52
53
- def get_modemversion (t91x_fota ):
54
- shadow = t91x_fota .fota .get_device (t91x_fota .device_id )
53
+ def get_modemversion (dut_fota ):
54
+ shadow = dut_fota .fota .get_device (dut_fota .device_id )
55
55
return shadow ["state" ]["reported" ]["device" ]["deviceInfo" ]["modemFirmware" ]
56
56
57
- def run_fota_resumption (t91x_fota , fota_type ):
57
+ def run_fota_resumption (dut_fota , fota_type ):
58
58
timeout_50_percent = APP_FOTA_TIMEOUT / 2
59
- t91x_fota .uart .wait_for_str ("50%" , timeout = timeout_50_percent )
59
+ dut_fota .uart .wait_for_str ("50%" , timeout = timeout_50_percent )
60
60
logger .debug (f"Testing fota resumption on disconnect for { fota_type } fota" )
61
61
62
62
patterns_lte_offline = ["network: Network connectivity lost" ]
63
63
patterns_lte_normal = ["network: Network connectivity established" ]
64
64
65
65
# LTE disconnect
66
- t91x_fota .uart .flush ()
67
- t91x_fota .uart .write ("zbus disconnect\r \n " )
68
- t91x_fota .uart .wait_for_str (patterns_lte_offline , timeout = 20 )
66
+ dut_fota .uart .flush ()
67
+ dut_fota .uart .write ("zbus disconnect\r \n " )
68
+ dut_fota .uart .wait_for_str (patterns_lte_offline , timeout = 20 )
69
69
70
70
# LTE reconnect
71
- t91x_fota .uart .flush ()
72
- t91x_fota .uart .write ("zbus connect\r \n " )
73
- t91x_fota .uart .wait_for_str (patterns_lte_normal , timeout = 120 )
74
-
75
- t91x_fota .uart .wait_for_str ("fota_download: Refuse fragment, restart with offset" )
76
- t91x_fota .uart .wait_for_str ("fota_download: Downloading from offset:" )
77
-
78
- def run_fota_reschedule (t91x_fota , fota_type ):
79
- t91x_fota .uart .wait_for_str (f"5%" , timeout = APP_FOTA_TIMEOUT )
71
+ dut_fota .uart .flush ()
72
+ dut_fota .uart .write ("zbus connect\r \n " )
73
+ dut_fota .uart .wait_for_str (patterns_lte_normal , timeout = 120 )
74
+ dut_fota .uart .wait_for_str ("fota_download: Refuse fragment, restart with offset" )
75
+ dut_fota .uart .wait_for_str ("fota_download: Downloading from offset:" )
76
+
77
+ def run_fota_reschedule (dut_fota , fota_type ):
78
+ dut_fota .uart .wait_for_str (f"5%" , timeout = APP_FOTA_TIMEOUT )
80
79
logger .debug (f"Cancelling FOTA, type: { fota_type } " )
81
80
82
- t91x_fota .fota .cancel_fota_job (t91x_fota .data ['job_id' ])
81
+ dut_fota .fota .cancel_fota_job (dut_fota .data ['job_id' ])
83
82
84
83
await_nrfcloud (
85
- functools .partial (t91x_fota .fota .get_fota_status , t91x_fota .data ['job_id' ]),
84
+ functools .partial (dut_fota .fota .get_fota_status , dut_fota .data ['job_id' ]),
86
85
"CANCELLED" ,
87
86
"FOTA status" ,
88
87
APP_FOTA_TIMEOUT
89
88
)
90
89
91
90
patterns_fota_cancel = ["Firmware download canceled" , "state_waiting_for_poll_request_entry" ]
92
91
93
- t91x_fota .uart .wait_for_str (patterns_fota_cancel , timeout = 180 )
92
+ dut_fota .uart .wait_for_str (patterns_fota_cancel , timeout = 180 )
94
93
95
- t91x_fota .data ['job_id' ] = t91x_fota .fota .create_fota_job (t91x_fota .device_id , t91x_fota .data ['bundle_id' ])
94
+ dut_fota .data ['job_id' ] = dut_fota .fota .create_fota_job (dut_fota .device_id , dut_fota .data ['bundle_id' ])
96
95
97
- logger .info (f"Rescheduled FOTA Job (ID: { t91x_fota .data ['job_id' ]} )" )
96
+ logger .info (f"Rescheduled FOTA Job (ID: { dut_fota .data ['job_id' ]} )" )
98
97
99
98
# Sleep a bit and trigger fota poll
100
99
for i in range (3 ):
101
100
try :
102
101
time .sleep (30 )
103
- t91x_fota .uart .write ("zbus button_press\r \n " )
104
- t91x_fota .uart .wait_for_str ("nrf_cloud_fota_poll: Starting FOTA download" )
102
+ dut_fota .uart .write ("zbus button_press\r \n " )
103
+ dut_fota .uart .wait_for_str ("nrf_cloud_fota_poll: Starting FOTA download" )
105
104
break
106
105
except AssertionError :
107
106
continue
108
107
else :
109
108
raise AssertionError (f"Fota update not available after { i } attempts" )
110
109
111
110
@pytest .fixture
112
- def run_fota_fixture (t91x_fota , hex_file , reschedule = False ):
111
+ def run_fota_fixture (dut_fota , hex_file , reschedule = False ):
113
112
def _run_fota (bundle_id = "" , fota_type = "app" , fotatimeout = APP_FOTA_TIMEOUT , new_version = TEST_APP_VERSION , reschedule = False ):
114
113
flash_device (os .path .abspath (hex_file ))
115
- t91x_fota .uart .xfactoryreset ()
116
- t91x_fota .uart .flush ()
114
+ dut_fota .uart .xfactoryreset ()
115
+ dut_fota .uart .flush ()
117
116
reset_device ()
118
- t91x_fota .uart .wait_for_str ("Connected to Cloud" )
117
+ dut_fota .uart .wait_for_str ("Connected to Cloud" )
119
118
120
119
time .sleep (60 )
121
120
122
121
if fota_type == "app" :
123
- bundle_id = t91x_fota .fota .upload_firmware (
122
+ bundle_id = dut_fota .fota .upload_firmware (
124
123
"nightly_test_app" ,
125
124
TEST_APP_BIN ,
126
125
TEST_APP_VERSION ,
@@ -130,39 +129,37 @@ def _run_fota(bundle_id="", fota_type="app", fotatimeout=APP_FOTA_TIMEOUT, new_v
130
129
logger .info (f"Uploaded file { TEST_APP_BIN } : bundleId: { bundle_id } " )
131
130
132
131
try :
133
- t91x_fota .data ['job_id' ] = t91x_fota .fota .create_fota_job (t91x_fota .device_id , bundle_id )
134
- t91x_fota .data ['bundle_id' ] = bundle_id
132
+ dut_fota .data ['job_id' ] = dut_fota .fota .create_fota_job (dut_fota .device_id , bundle_id )
133
+ dut_fota .data ['bundle_id' ] = bundle_id
135
134
except NRFCloudFOTAError as e :
136
135
pytest .skip (f"FOTA create_job REST API error: { e } " )
137
- logger .info (f"Created FOTA Job (ID: { t91x_fota .data ['job_id' ]} )" )
136
+ logger .info (f"Created FOTA Job (ID: { dut_fota .data ['job_id' ]} )" )
138
137
139
138
# Sleep a bit and trigger fota poll
140
139
for i in range (3 ):
141
140
try :
142
141
time .sleep (10 )
143
- t91x_fota .uart .write ("zbus button_press\r \n " )
144
- t91x_fota .uart .wait_for_str ("nrf_cloud_fota_poll: Starting FOTA download" )
142
+ dut_fota .uart .write ("zbus button_press\r \n " )
143
+ dut_fota .uart .wait_for_str ("nrf_cloud_fota_poll: Starting FOTA download" )
145
144
break
146
145
except AssertionError :
147
146
continue
148
147
else :
149
148
raise AssertionError (f"Fota update not available after { i } attempts" )
150
149
151
150
if reschedule :
152
- run_fota_reschedule (t91x_fota , fota_type )
151
+ run_fota_reschedule (dut_fota , fota_type )
153
152
154
153
if fota_type == "app" :
155
- run_fota_resumption (t91x_fota , "app" )
156
-
154
+ run_fota_resumption (dut_fota , "app" )
157
155
await_nrfcloud (
158
- functools .partial (t91x_fota .fota .get_fota_status , t91x_fota .data ['job_id' ]),
156
+ functools .partial (dut_fota .fota .get_fota_status , dut_fota .data ['job_id' ]),
159
157
"IN_PROGRESS" ,
160
158
"FOTA status" ,
161
159
fotatimeout
162
160
)
163
-
164
161
await_nrfcloud (
165
- functools .partial (t91x_fota .fota .get_fota_status , t91x_fota .data ['job_id' ]),
162
+ functools .partial (dut_fota .fota .get_fota_status , dut_fota .data ['job_id' ]),
166
163
"COMPLETED" ,
167
164
"FOTA status" ,
168
165
fotatimeout
@@ -171,14 +168,14 @@ def _run_fota(bundle_id="", fota_type="app", fotatimeout=APP_FOTA_TIMEOUT, new_v
171
168
try :
172
169
if fota_type == "app" :
173
170
await_nrfcloud (
174
- functools .partial (get_appversion , t91x_fota ),
171
+ functools .partial (get_appversion , dut_fota ),
175
172
new_version ,
176
173
"appVersion" ,
177
174
DEVICE_MSG_TIMEOUT
178
175
)
179
176
else :
180
177
await_nrfcloud (
181
- functools .partial (get_modemversion , t91x_fota ),
178
+ functools .partial (get_modemversion , dut_fota ),
182
179
new_version ,
183
180
"modemFirmware" ,
184
181
DEVICE_MSG_TIMEOUT
0 commit comments