vincent/aptly
1
0
Fork 0
mirror of https://github.com/aptly-dev/aptly.git synced 2026-04-19 19:28:22 +00:00
Code Issues Projects Releases Wiki Activity
Files
ff1557afeefc4699f7146cdb943c39ed1a052d42
aptly/system/t12_api/snapshots.py
André Roth 57e2c5c670 api tests: use check_task_fail
2024-07-24 21:19:47 +02:00

396 lines
16 KiB
Python
Raw Blame History

from api_lib import APITest
from .publish import DefaultSigningOptions
class SnapshotsAPITestCreateShowEmpty(APITest):
"""
GET /api/snapshots/:name, POST /api/snapshots, GET /api/snapshots/:name/packages
"""
def check(self):
snapshot_name = self.random_name()
snapshot_desc = {'Description': 'fun snapshot',
'Name': snapshot_name}
# create empty snapshot
task = self.post_task("/api/snapshots", json=snapshot_desc)
self.check_task(task)
self.check_subset(snapshot_desc, self.get("/api/snapshots/" + snapshot_name).json())
self.check_equal(self.get("/api/snapshots/" + snapshot_name).status_code, 200)
resp = self.get("/api/snapshots/" + snapshot_name + "/packages")
self.check_equal(resp.status_code, 200)
self.check_equal(resp.json(), [])
self.check_equal(self.get("/api/snapshots/" + self.random_name()).status_code, 404)
# create snapshot with duplicate name
task = self.post_task("/api/snapshots", json=snapshot_desc)
self.check_task_fail(task)
class SnapshotsAPITestCreateFromRefs(APITest):
"""
GET /api/snapshots/:name, POST /api/snapshots, GET /api/snapshots/:name/packages,
GET /api/snapshots
"""
def check(self):
snapshot_name = self.random_name()
snapshot_desc = {'Description': 'fun snapshot',
'Name': snapshot_name,
'SourceSnapshots': [self.random_name()]}
# creating snapshot from missing source snapshot
resp = self.post("/api/snapshots", json=snapshot_desc)
self.check_equal(resp.status_code, 404)
# create empty snapshot
empty_snapshot_name = self.random_name()
task = self.post_task("/api/snapshots", json={"Name": empty_snapshot_name})
self.check_task(task)
self.check_equal(
self.get("/api/snapshots/" + empty_snapshot_name).json()['Description'], "Created as empty"
)
# create and upload package to repo to register package in DB
repo_name = self.random_name()
self.check_equal(self.post("/api/repos", json={"Name": repo_name}).status_code, 201)
d = self.random_name()
self.check_equal(self.upload("/api/files/" + d,
"libboost-program-options-dev_1.49.0.1_i386.deb").status_code, 200)
task = self.post_task("/api/repos/" + repo_name + "/file/" + d)
self.check_task(task)
# create snapshot with empty snapshot as source and package
snapshot = snapshot_desc.copy()
snapshot['PackageRefs'] = ["Pi386 libboost-program-options-dev 1.49.0.1 918d2f433384e378"]
snapshot['SourceSnapshots'] = [empty_snapshot_name]
task = self.post_task("/api/snapshots", json=snapshot)
self.check_task(task)
snapshot.pop('SourceSnapshots')
snapshot.pop('PackageRefs')
resp = self.get("/api/snapshots/" + snapshot_name)
self.check_equal(resp.status_code, 200)
self.check_subset(snapshot, resp.json())
self.check_subset(snapshot, self.get("/api/snapshots/" + snapshot_name).json())
resp = self.get("/api/snapshots/" + snapshot_name + "/packages")
self.check_equal(resp.status_code, 200)
self.check_equal(resp.json(), ["Pi386 libboost-program-options-dev 1.49.0.1 918d2f433384e378"])
# create snapshot with unreferenced package
task = self.post_task("/api/snapshots", json={
"Name": self.random_name(),
"PackageRefs": ["Pi386 libboost-program-options-dev 1.49.0.1 918d2f433384e378", "Pamd64 no-such-package 1.2 91"]})
self.check_task_fail(task)
# list snapshots
resp = self.get("/api/snapshots", params={"sort": "time"})
self.check_equal(resp.status_code, 200)
self.check_equal([s["Name"] for s in resp.json() if s["Name"] in [empty_snapshot_name, snapshot_name]],
[empty_snapshot_name, snapshot_name])
class SnapshotsAPITestCreateFromRepo(APITest):
"""
POST /api/repos, POST /api/repos/:name/snapshots, GET /api/snapshots/:name
"""
def check(self):
repo_name = self.random_name()
snapshot_name = self.random_name()
self.check_equal(self.post("/api/repos", json={"Name": repo_name}).status_code, 201)
task = self.post_task("/api/repos/" + repo_name + '/snapshots', json={'Name': snapshot_name})
self.check_task(task)
self.check_equal([],
self.get("/api/snapshots/" + snapshot_name + "/packages", params={"format": "details"}).json())
snapshot_name = self.random_name()
d = self.random_name()
self.check_equal(self.upload("/api/files/" + d,
"libboost-program-options-dev_1.49.0.1_i386.deb").status_code, 200)
task = self.post_task("/api/repos/" + repo_name + "/file/" + d)
self.check_task(task)
task = self.post_task("/api/repos/" + repo_name + '/snapshots', json={'Name': snapshot_name})
self.check_task(task)
self.check_equal(self.get("/api/snapshots/" + snapshot_name).status_code, 200)
self.check_subset({'Architecture': 'i386',
'Package': 'libboost-program-options-dev',
'Version': '1.49.0.1',
'FilesHash': '918d2f433384e378'},
self.get("/api/snapshots/" + snapshot_name + "/packages", params={"format": "details"}).json()[0])
self.check_subset({'Architecture': 'i386',
'Package': 'libboost-program-options-dev',
'Version': '1.49.0.1',
'FilesHash': '918d2f433384e378'},
self.get("/api/snapshots/" + snapshot_name + "/packages",
params={"format": "details", "q": "Version (> 0.6.1-1.4)"}).json()[0])
# duplicate snapshot name
task = self.post_task("/api/repos/" + repo_name + '/snapshots', json={'Name': snapshot_name})
self.check_task_fail(task)
class SnapshotsAPITestCreateUpdate(APITest):
"""
POST /api/snapshots, PUT /api/snapshots/:name, GET /api/snapshots/:name
"""
def check(self):
snapshot_name = self.random_name()
snapshot_desc = {'Description': 'fun snapshot',
'Name': snapshot_name}
task = self.post_task("/api/snapshots", json=snapshot_desc)
self.check_task(task)
new_snapshot_name = self.random_name()
task = self.put_task("/api/snapshots/" + snapshot_name, json={'Name': new_snapshot_name,
'Description': 'New description'})
self.check_task(task)
resp = self.get("/api/snapshots/" + new_snapshot_name)
self.check_equal(resp.status_code, 200)
self.check_subset({"Name": new_snapshot_name,
"Description": "New description"}, resp.json())
# duplicate name
task = self.put_task("/api/snapshots/" + new_snapshot_name, json={'Name': new_snapshot_name,
'Description': 'New description'})
self.check_task_fail(task)
# missing snapshot
resp = self.put("/api/snapshots/" + snapshot_name, json={})
self.check_equal(resp.status_code, 404)
class SnapshotsAPITestCreateDelete(APITest):
"""
POST /api/snapshots, DELETE /api/snapshots/:name, GET /api/snapshots/:name
"""
def check(self):
snapshot_name = self.random_name()
snapshot_desc = {'Description': 'fun snapshot',
'Name': snapshot_name}
# deleting unreferenced snapshot
task = self.post_task("/api/snapshots", json=snapshot_desc)
self.check_task(task)
task = self.delete_task("/api/snapshots/" + snapshot_name)
self.check_task(task)
self.check_equal(self.get("/api/snapshots/" + snapshot_name).status_code, 404)
# deleting referenced snapshot
snap1, snap2 = self.random_name(), self.random_name()
task = self.post_task("/api/snapshots", json={"Name": snap1})
self.check_task(task)
self.check_equal(
self.post_task(
"/api/snapshots", json={"Name": snap2, "SourceSnapshots": [snap1]}
).json()['State'], 2
)
task = self.delete_task("/api/snapshots/" + snap1)
self.check_task_fail(task)
self.check_equal(self.get("/api/snapshots/" + snap1).status_code, 200)
task = self.delete_task("/api/snapshots/" + snap1, params={"force": "1"})
self.check_task(task)
self.check_equal(self.get("/api/snapshots/" + snap1).status_code, 404)
# deleting published snapshot
task = self.post_task(
"/api/publish",
json={
"SourceKind": "snapshot",
"Distribution": "trusty",
"Architectures": ["i386"],
"Sources": [{"Name": snap2}],
"Signing": DefaultSigningOptions,
}
)
self.check_task(task)
task = self.delete_task("/api/snapshots/" + snap2)
self.check_task_fail(task)
task = self.delete_task("/api/snapshots/" + snap2, params={"force": "1"})
self.check_task_fail(task)
class SnapshotsAPITestSearch(APITest):
"""
POST /api/snapshots, GET /api/snapshots?sort=name, GET /api/snapshots/:name
"""
def check(self):
repo_name = self.random_name()
self.check_equal(self.post("/api/repos", json={"Name": repo_name}).status_code, 201)
d = self.random_name()
snapshot_name = self.random_name()
self.check_equal(self.upload("/api/files/" + d,
"libboost-program-options-dev_1.49.0.1_i386.deb").status_code, 200)
task = self.post_task("/api/repos/" + repo_name + "/file/" + d)
self.check_task(task)
task = self.post_task("/api/repos/" + repo_name + '/snapshots', json={'Name': snapshot_name})
self.check_task(task)
resp = self.get("/api/snapshots/" + snapshot_name + "/packages",
params={"q": "libboost-program-options-dev", "format": "details"})
self.check_equal(resp.status_code, 200)
self.check_equal(len(resp.json()), 1)
self.check_equal(resp.json()[0]["Package"], "libboost-program-options-dev")
resp = self.get("/api/snapshots/" + snapshot_name + "/packages")
self.check_equal(resp.status_code, 200)
self.check_equal(len(resp.json()), 1)
self.check_equal(resp.json(), ["Pi386 libboost-program-options-dev 1.49.0.1 918d2f433384e378"])
class SnapshotsAPITestDiff(APITest):
"""
GET /api/snapshot/:name/diff/:name2
"""
def check(self):
repos = [self.random_name() for x in range(2)]
snapshots = [self.random_name() for x in range(2)]
for repo_name in repos:
self.check_equal(self.post("/api/repos", json={"Name": repo_name}).status_code, 201)
d = self.random_name()
self.check_equal(self.upload("/api/files/" + d,
"libboost-program-options-dev_1.49.0.1_i386.deb").status_code, 200)
task = self.post_task("/api/repos/" + repos[-1] + "/file/" + d)
self.check_task(task)
task = self.post_task("/api/repos/" + repos[-1] + '/snapshots', json={'Name': snapshots[0]})
self.check_task(task)
task = self.post_task("/api/snapshots", json={'Name': snapshots[1]})
self.check_task(task)
resp = self.get("/api/snapshots/" + snapshots[0] + "/diff/" + snapshots[1])
self.check_equal(resp.status_code, 200)
self.check_equal(resp.json(), [{'Left': 'Pi386 libboost-program-options-dev 1.49.0.1 918d2f433384e378',
'Right': None}])
resp = self.get("/api/snapshots/" + snapshots[1] + "/diff/" + snapshots[0])
self.check_equal(resp.status_code, 200)
self.check_equal(resp.json(), [{'Right': 'Pi386 libboost-program-options-dev 1.49.0.1 918d2f433384e378',
'Left': None}])
resp = self.get("/api/snapshots/" + snapshots[0] + "/diff/" + snapshots[0])
self.check_equal(resp.status_code, 200)
self.check_equal(resp.json(), [])
resp = self.get("/api/snapshots/" + snapshots[1] + "/diff/" + snapshots[1])
self.check_equal(resp.status_code, 200)
self.check_equal(resp.json(), [])
class SnapshotsAPITestMerge(APITest):
"""
POST /api/snapshots, GET /api/snapshots/merge, GET /api/snapshots/:name, DELETE /api/snapshots/:name
"""
def check(self):
sources = [
{"Description": "fun snapshot", "Name": self.random_name()}
for _ in range(2)
]
# create source snapshots
for source in sources:
task = self.post_task("/api/snapshots", json=source)
self.check_task(task)
# create merge snapshot
merged_name = self.random_name()
task = self.post_task(
"/api/snapshots/merge",
json={
"Destination": merged_name,
"Sources": [source["Name"] for source in sources],
},
)
self.check_task(task)
# check merge snapshot
resp = self.get(f"/api/snapshots/{merged_name}")
self.check_equal(resp.status_code, 200)
source_list = ", ".join(f"'{source['Name']}'" for source in sources)
self.check_subset(
{
"Name": merged_name,
"Description": f"Merged from sources: {source_list}",
},
resp.json(),
)
# remove merge snapshot
task = self.delete_task(f"/api/snapshots/{merged_name}")
self.check_task(task)
# create merge snapshot without sources
merged_name = self.random_name()
resp = self.post(
"/api/snapshots/merge", json={"Destination": merged_name, "Sources": []}
)
self.check_equal(resp.status_code, 400)
self.check_equal(
resp.json()["error"], "At least one source snapshot is required"
)
self.check_equal(self.get(f"/api/snapshots/{merged_name}").status_code, 404)
# create merge snapshot with non-existing source
merged_name = self.random_name()
non_existing_source = self.random_name()
resp = self.post(
"/api/snapshots/merge",
json={"Destination": merged_name, "Sources": [non_existing_source]},
)
self.check_equal(
resp.json()["error"], f"snapshot with name {non_existing_source} not found"
)
self.check_equal(resp.status_code, 404)
self.check_equal(self.get(f"/api/snapshots/{merged_name}").status_code, 404)
# create merge snapshot with used name
merged_name = sources[0]["Name"]
resp = self.post(
"/api/snapshots/merge",
json={"Destination": merged_name, "Sources": [source["Name"] for source in sources]},
)
self.check_equal(
resp.json()["error"],
f"unable to create snapshot: snapshot with name {sources[0]['Name']} already exists",
)
self.check_equal(resp.status_code, 500)
# create merge snapshot with "latest" and "no-remove" flags (should fail)
merged_name = self.random_name()
resp = self.post(
"/api/snapshots/merge",
json={
"Destination": merged_name,
"Sources": [source["Name"] for source in sources],
},
params={"latest": "1", "no-remove": "1"},
)
self.check_equal(
resp.json()["error"], "no-remove and latest are mutually exclusive"
)
self.check_equal(resp.status_code, 400)
Reference in New Issue View Git Blame Copy Permalink