1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
|
import sys
import time
import os
import subprocess
import requests
from retry import retry
from test_log_generator import create_example_logs
"""
Validate end to end behaviour for the log request operation.
When we send a log request ( for all text with, 300 lines) from cumulocity to device and wait for some time.
Then sm mapper receives the request and sends the response
Validate if the response contains the log file for number of lines.
If number of lines are greater than 300 then pass the test
Else stop and cleanup the operation by sending operation failed message.
"""
from environment_c8y import EnvironmentC8y
class LogRequestVerifyNumberOfLines(EnvironmentC8y):
operation_id = None
def setup(self):
super().setup()
self.create_logs_for_test()
# Start c8y logfile request service
log_file_daemon = self.startProcess(
command=self.sudo,
arguments=[self.systemctl, "start", "c8y-log-plugin.service"],
stdouterr="log_file_daemon",
)
self.addCleanupFunction(self.cleanup_logs)
def execute(self):
log_file_request_payload = {
"dateFrom": "2022-06-04T12:55:49+0530",
"dateTo": "2022-06-06T13:55:49+0530",
"logFile": "software-management",
"searchText": "",
"maximumLines": 300,
}
self.operation_id = self.cumulocity.trigger_log_request(
log_file_request_payload, self.project.deviceid
)
def validate(self):
self.assertThat("True == value", value=self.wait_until_logs_retrieved())
@retry(Exception, tries=20, delay=1)
def wait_until_logs_retrieved(self):
log_file = self.cumulocity.retrieve_log_file(self.operation_id)
if len(log_file) != 0:
return self.download_file_and_verify_number_of_lines(log_file)
else:
raise Exception("retry")
def create_logs_for_test(self):
# remove if there are any old files
rm_logs = self.startProcess(
command=self.sudo,
arguments=["rm", "-rf", "/tmp/sw_logs"],
stdouterr="rm_logs",
)
# create example logs
create_example_logs()
move_logs = self.startProcess(
command=self.sudo,
arguments=["sh", "-c", "mv /tmp/sw_logs/* /var/log/tedge/agent/"],
stdouterr="move_logs",
)
def download_file_and_verify_number_of_lines(self, url):
get_response = requests.get(
url, auth=(self.project.c8yusername, self.project.c8ypass), stream=False
)
nlines = len(get_response.content.decode("utf-8").split("\n")[:-1])
self.log.info("nlines %s", nlines)
# The log lines are concatenated from 3 different log files, so there will be 3 extra lines.
if nlines == 302:
return True
else:
return False
def cleanup_logs(self):
rm_logs = self.startProcess(
command=self.sudo,
arguments=["sh", "-c", "rm -rf /var/log/tedge/agent/example-*"],
stdouterr="rm_logs",
)
rm_tmp_logs = self.startProcess(
command=self.sudo,
arguments=["rm", "-rf", "/tmp/sw_logs"],
stdouterr="rm_tmp_logs",
)
if self.getOutcome().isFailure():
log = self.startProcess(
command=self.sudo,
arguments=[
self.tedge,
"mqtt",
"pub",
"c8y/s/us",
"502,c8y_LogfileRequest",
],
stdouterr="send_failed",
)
log_file_daemon = self.startProcess(
command=self.sudo,
arguments=[self.systemctl, "stop", "c8y-log-plugin.service"],
stdouterr="log_file_daemon",
)
|
