Expand source code
@pytest.mark.p0
@pytest.mark.support_bundle
class TestSupportBundle:
@pytest.mark.dependency(name="create support bundle")
def test_create(self, api_client, unique_name, support_bundle_state):
code, data = api_client.supportbundle.create(unique_name)
assert 201 == code, (code, data)
support_bundle_state.uid = data['metadata']['name']
@pytest.mark.dependency(name="get support bundle", depends=["create support bundle"])
def test_get(self, api_client, support_bundle_state):
code, data = api_client.supportbundle.get(support_bundle_state.uid)
assert 200 == code, (code, data)
@pytest.mark.dependency(name="donwnload support bundle", depends=["get support bundle"])
def test_download(self, api_client, support_bundle_state, wait_timeout):
endtime = datetime.now() + timedelta(seconds=wait_timeout)
while endtime > datetime.now():
code, data = api_client.supportbundle.get(support_bundle_state.uid)
if 100 == data.get('status', {}).get('progress', 0):
break
sleep(5)
else:
raise AssertionError(
f"Failed to wait supportbundle ready with {wait_timeout} timed out\n"
f"Still got {code} in {data}"
)
code, ctx = api_client.supportbundle.download(support_bundle_state.uid)
assert 200 == code, (code, ctx)
with ZipFile(BytesIO(ctx), 'r') as zf:
files = zf.namelist()
assert 0 != len(files)
support_bundle_state.files = files
support_bundle_state.fio.write(ctx)
support_bundle_state.fio.seek(0)
@pytest.mark.dependency(depends=["donwnload support bundle"])
def test_logfile_exists(self, support_bundle_state):
patterns = [r"^.*/logs/cattle-fleet-local-system/fleet-agent-.*/fleet-agent.log",
r"^.*/logs/cattle-fleet-system/fleet-controller-.*/fleet-controller.log",
r"^.*/logs/cattle-fleet-system/gitjob-.*/gitjob.log"]
matches = []
for f in support_bundle_state.files:
for pattern in patterns:
matches.extend([f] if re.match(pattern, f) else [])
assert len(matches) == len(patterns), (
f"Some file(s) not found, files: {matches}\npatterns: {patterns}"
)
@pytest.mark.dependency(depends=["get support bundle"])
def test_delete(self, api_client, support_bundle_state):
code, data = api_client.supportbundle.delete(support_bundle_state.uid)
# ???: Downloaded support bundle will be deleted automatically
assert 404 == code, (code, data)