vincent/aptly
1
0
Fork 0
mirror of https://github.com/aptly-dev/aptly.git synced 2026-06-16 07:10:54 +00:00
Code Issues Projects Releases Wiki Activity

New upstream version 1.5.0+ds1

Browse Source
This commit is contained in:
Roland Mas
2023-01-02 14:19:29 +01:00
parent 29e4ea6ec0
commit 5c4f97f88e
324 changed files with 15360 additions and 6668 deletions
Download Patch File Download Diff File Expand all files Collapse all files
system/t12_api/db.py
+14
View File
@@ -0,0 +1,14 @@
from api_lib import APITest
class DbAPITestCleanup(APITest):
"""
POST /db/cleanup
"""
def check(self):
resp = self.post_task(
"/api/db/cleanup"
)
self.check_equal(resp.status_code, 200)
system/t12_api/gpg.py
+54
View File
@@ -0,0 +1,54 @@
import inspect
import os
import subprocess
import tempfile
from api_lib import APITest
def check_gpgkey_exists(gpg_key, keyring):
subprocess.check_call([
"gpg", "--no-default-keyring",
"--keyring", keyring,
"--fingerprint", gpg_key,
])
class GPGAPITestAddKey(APITest):
"""
POST /gpg/key
"""
def check(self):
with tempfile.NamedTemporaryFile(suffix=".pub") as keyring:
gpgkeyid = "9E3E53F19C7DE460"
resp = self.post("/api/gpg/key", json={
"Keyserver": "keyserver.ubuntu.com",
"Keyring": keyring.name,
"GpgKeyID": gpgkeyid
})
self.check_equal(resp.status_code, 200)
check_gpgkey_exists(gpgkeyid, keyring.name)
class GPGAPITestAddKeyArmor(APITest):
"""
POST /gpg/key
"""
def check(self):
keyfile = os.path.join(os.path.dirname(inspect.getsourcefile(APITest)),
"files") + "/launchpad.key"
gpgkeyid = "3B1F56C0"
with open(keyfile, 'r') as keyf:
gpgkeyarmor = keyf.read()
with tempfile.NamedTemporaryFile(suffix=".pub") as keyring:
resp = self.post("/api/gpg/key", json={
"Keyring": keyring.name,
"GpgKeyArmor": gpgkeyarmor
})
self.check_equal(resp.status_code, 200)
check_gpgkey_exists(gpgkeyid, keyring.name)
system/t12_api/graph.py
+8 -4
View File
@@ -7,19 +7,21 @@ class GraphAPITest(APITest):
GET /graph.:ext
"""
requiresDot = True
def check(self):
resp = self.get("/api/graph.png")
self.check_equal(resp.headers["Content-Type"], "image/png")
self.check_equal(resp.content[:4], '\x89PNG')
self.check_equal(resp.content[:4], b'\x89PNG')
self.check_equal(self.post("/api/repos", json={"Name": "xyz", "Comment": "fun repo"}).status_code, 201)
resp = self.get("/api/graph.svg")
self.check_equal(resp.headers["Content-Type"], "image/svg+xml")
self.check_equal(resp.content[:4], '<?xm')
self.check_equal(resp.content[:4], b'<?xm')
resp = self.get("/api/graph.dot")
self.check_equal(resp.headers["Content-Type"], "text/plain; charset=utf-8")
self.check_equal(resp.content[:13], 'digraph aptly')
self.check_equal(resp.content[:13], b'digraph aptly')
# basic test of layout:
# horizontal should be wider than vertical
@@ -44,4 +46,6 @@ class GraphAPITest(APITest):
# remove the repos again
for repo in tempRepos:
self.check_equal(self.delete("/api/repos/" + repo, params={"force": "1"}).status_code, 200)
self.check_equal(self.delete_task(
"/api/repos/" + repo, params={"force": "1"}).json()['State'], 2
)
system/t12_api/metrics.py
+26
View File
@@ -0,0 +1,26 @@
from api_lib import APITest
class MetricsEnabledAPITest(APITest):
"""
GET /metrics
"""
def check(self):
resp = self.get("/api/metrics")
self.check_equal(resp.status_code, 200)
apiRequestsInFlightGauge = "# TYPE aptly_api_http_requests_in_flight gauge"
self.check_in(apiRequestsInFlightGauge, resp.text)
apiRequestsTotalCounter = "# TYPE aptly_api_http_requests_total counter"
self.check_in(apiRequestsTotalCounter, resp.text)
apiRequestSizeSummary = "# TYPE aptly_api_http_request_size_bytes summary"
self.check_in(apiRequestSizeSummary, resp.text)
apiResponseSizeSummary = "# TYPE aptly_api_http_response_size_bytes summary"
self.check_in(apiResponseSizeSummary, resp.text)
apiRequestsDurationSummary = "# TYPE aptly_api_http_request_duration_seconds summary"
self.check_in(apiRequestsDurationSummary, resp.text)
system/t12_api/mirrors.py
+117
View File
@@ -0,0 +1,117 @@
from api_lib import APITest
class MirrorsAPITestCreateShow(APITest):
"""
POST /api/mirrors, GET /api/mirrors/:name/packages
"""
def check(self):
mirror_name = self.random_name()
mirror_desc = {'Name': mirror_name,
'ArchiveURL': 'http://security.debian.org/debian-security/',
'Architectures': ['amd64'],
'Components': ['main'],
'Distribution': 'buster/updates'}
resp = self.post("/api/mirrors", json=mirror_desc)
self.check_equal(resp.status_code, 400)
self.check_equal({
'error': 'unable to fetch mirror: verification of detached signature failed: exit status 2',
}, resp.json())
mirror_desc['IgnoreSignatures'] = True
resp = self.post("/api/mirrors", json=mirror_desc)
self.check_equal(resp.status_code, 201)
resp = self.get("/api/mirrors/" + mirror_name)
self.check_equal(resp.status_code, 200)
self.check_subset({'Name': mirror_name,
'ArchiveRoot': 'http://security.debian.org/debian-security/',
'Architectures': ['amd64'],
'Components': ['main'],
'Distribution': 'buster/updates'}, resp.json())
resp = self.get("/api/mirrors/" + mirror_desc["Name"] + "/packages")
self.check_equal(resp.status_code, 404)
class MirrorsAPITestCreateUpdate(APITest):
"""
POST /api/mirrors, PUT /api/mirrors/:name, GET /api/mirrors/:name/packages
"""
def check(self):
mirror_name = self.random_name()
mirror_desc = {'Name': mirror_name,
'ArchiveURL': 'https://packagecloud.io/varnishcache/varnish30/debian/',
'Distribution': 'wheezy',
'Components': ['main']}
mirror_desc['IgnoreSignatures'] = True
resp = self.post("/api/mirrors", json=mirror_desc)
self.check_equal(resp.status_code, 201)
resp = self.get("/api/mirrors/" + mirror_name + "/packages")
self.check_equal(resp.status_code, 404)
mirror_desc["Name"] = self.random_name()
resp = self.put_task("/api/mirrors/" + mirror_name, json=mirror_desc)
self.check_equal(resp.json()["State"], 2)
_id = resp.json()['ID']
resp = self.get("/api/tasks/" + str(_id) + "/detail")
self.check_equal(resp.status_code, 200)
self.check_equal(resp.json()['RemainingDownloadSize'], 0)
self.check_equal(resp.json()['RemainingNumberOfPackages'], 0)
resp = self.get("/api/mirrors/" + mirror_desc["Name"])
self.check_equal(resp.status_code, 200)
self.check_subset({'Name': mirror_desc["Name"],
'ArchiveRoot': 'https://packagecloud.io/varnishcache/varnish30/debian/',
'Distribution': 'wheezy'}, resp.json())
resp = self.get("/api/mirrors/" + mirror_desc["Name"] + "/packages")
self.check_equal(resp.status_code, 200)
class MirrorsAPITestCreateDelete(APITest):
"""
POST /api/mirrors, DELETE /api/mirrors/:name
"""
def check(self):
mirror_name = self.random_name()
mirror_desc = {'Name': mirror_name,
'ArchiveURL': 'https://packagecloud.io/varnishcache/varnish30/debian/',
'IgnoreSignatures': True,
'Distribution': 'wheezy',
'Components': ['main']}
resp = self.post("/api/mirrors", json=mirror_desc)
self.check_equal(resp.status_code, 201)
resp = self.delete_task("/api/mirrors/" + mirror_name)
self.check_equal(resp.json()['State'], 2)
class MirrorsAPITestCreateList(APITest):
"""
GET /api/mirrors, POST /api/mirrors, GET /api/mirrors
"""
def check(self):
resp = self.get("/api/mirrors")
self.check_equal(resp.status_code, 200)
count = len(resp.json())
mirror_name = self.random_name()
mirror_desc = {'Name': mirror_name,
'ArchiveURL': 'https://packagecloud.io/varnishcache/varnish30/debian/',
'IgnoreSignatures': True,
'Distribution': 'wheezy',
'Components': ['main']}
resp = self.post("/api/mirrors", json=mirror_desc)
self.check_equal(resp.status_code, 201)
resp = self.get("/api/mirrors")
self.check_equal(resp.status_code, 200)
self.check_equal(len(resp.json()), count + 1)
system/t12_api/packages.py
+8 -5
View File
@@ -1,4 +1,7 @@
import urllib
import urllib.error
import urllib.parse
import urllib.request
from api_lib import APITest
@@ -15,11 +18,11 @@ class PackagesAPITestShow(APITest):
self.check_equal(self.upload("/api/files/" + d,
"pyspi_0.6.1-1.3.dsc", "pyspi_0.6.1-1.3.diff.gz", "pyspi_0.6.1.orig.tar.gz").status_code, 200)
resp = self.post("/api/repos/" + repo_name + "/file/" + d)
self.check_equal(resp.status_code, 200)
resp = self.post_task("/api/repos/" + repo_name + "/file/" + d)
self.check_equal(resp.json()['State'], 2)
# get information about package
resp = self.get("/api/packages/" + urllib.quote('Psource pyspi 0.6.1-1.3 3a8b37cbd9a3559e'))
resp = self.get("/api/packages/" + urllib.parse.quote('Psource pyspi 0.6.1-1.3 3a8b37cbd9a3559e'))
self.check_equal(resp.status_code, 200)
self.check_equal(resp.json(), {
'Architecture': 'any',
@@ -40,5 +43,5 @@ class PackagesAPITestShow(APITest):
'Vcs-Svn': 'svn://svn.tribulaciones.org/srv/svn/pyspi/trunk',
'Version': '0.6.1-1.3'})
resp = self.get("/api/packages/" + urllib.quote('Pamd64 no-such-package 1.0 3a8b37cbd9a3559e'))
resp = self.get("/api/packages/" + urllib.parse.quote('Pamd64 no-such-package 1.0 3a8b37cbd9a3559e'))
self.check_equal(resp.status_code, 404)
system/t12_api/publish.py
+281 -188
View File
@@ -17,7 +17,8 @@ class PublishAPITestRepo(APITest):
def check(self):
repo_name = self.random_name()
self.check_equal(self.post("/api/repos", json={"Name": repo_name, "DefaultDistribution": "wheezy"}).status_code, 201)
self.check_equal(self.post(
"/api/repos", json={"Name": repo_name, "DefaultDistribution": "wheezy"}).status_code, 201)
d = self.random_name()
self.check_equal(self.upload("/api/files/" + d,
@@ -25,16 +26,18 @@ class PublishAPITestRepo(APITest):
"pyspi_0.6.1-1.3.diff.gz", "pyspi_0.6.1.orig.tar.gz",
"pyspi-0.6.1-1.3.stripped.dsc").status_code, 200)
self.check_equal(self.post("/api/repos/" + repo_name + "/file/" + d).status_code, 200)
self.check_equal(self.post_task("/api/repos/" + repo_name + "/file/" + d).json()['State'], 2)
# publishing under prefix, default distribution
prefix = self.random_name()
resp = self.post("/api/publish/" + prefix,
json={
"SourceKind": "local",
"Sources": [{"Name": repo_name}],
"Signing": DefaultSigningOptions,
})
resp = self.post_task(
"/api/publish/" + prefix,
json={
"SourceKind": "local",
"Sources": [{"Name": repo_name}],
"Signing": DefaultSigningOptions,
}
)
repo_expected = {
'AcquireByHash': False,
'Architectures': ['i386', 'source'],
@@ -43,35 +46,43 @@ class PublishAPITestRepo(APITest):
'Origin': '',
'NotAutomatic': '',
'ButAutomaticUpgrades': '',
'Path': prefix + '/' + 'wheezy',
'Prefix': prefix,
'SkipContents': False,
'SourceKind': 'local',
'Sources': [{'Component': 'main', 'Name': repo_name}],
'Storage': ''}
'Storage': '',
'Suite': ''}
self.check_equal(resp.status_code, 201)
self.check_equal(resp.json(), repo_expected)
self.check_equal(resp.json()['State'], 2)
all_repos = self.get("/api/publish")
self.check_equal(all_repos.status_code, 200)
self.check_in(repo_expected, all_repos.json())
self.check_exists("public/" + prefix + "/dists/wheezy/Release")
self.check_exists("public/" + prefix + "/dists/wheezy/main/binary-i386/Packages")
self.check_exists("public/" + prefix + "/dists/wheezy/main/Contents-i386.gz")
self.check_exists("public/" + prefix + "/dists/wheezy/main/source/Sources")
self.check_exists("public/" + prefix + "/pool/main/b/boost-defaults/libboost-program-options-dev_1.49.0.1_i386.deb")
self.check_exists("public/" + prefix +
"/dists/wheezy/main/binary-i386/Packages")
self.check_exists("public/" + prefix +
"/dists/wheezy/main/Contents-i386.gz")
self.check_exists("public/" + prefix +
"/dists/wheezy/main/source/Sources")
self.check_exists(
"public/" + prefix + "/pool/main/b/boost-defaults/libboost-program-options-dev_1.49.0.1_i386.deb")
# publishing under root, custom distribution, architectures
distribution = self.random_name()
resp = self.post("/api/publish/:.",
json={
"SourceKind": "local",
"Sources": [{"Name": repo_name}],
"Signing": DefaultSigningOptions,
"Distribution": distribution,
"Architectures": ["i386", "amd64"],
})
resp = self.post_task(
"/api/publish/:.",
json={
"SourceKind": "local",
"Sources": [{"Name": repo_name}],
"Signing": DefaultSigningOptions,
"Distribution": distribution,
"Architectures": ["i386", "amd64"],
}
)
self.check_equal(resp.json()['State'], 2)
repo2_expected = {
'AcquireByHash': False,
'Architectures': ['amd64', 'i386'],
@@ -80,20 +91,28 @@ class PublishAPITestRepo(APITest):
'Origin': '',
'NotAutomatic': '',
'ButAutomaticUpgrades': '',
'Path': './' + distribution,
'Prefix': ".",
'SkipContents': False,
'SourceKind': 'local',
'Sources': [{'Component': 'main', 'Name': repo_name}],
'Storage': ''}
self.check_equal(resp.status_code, 201)
self.check_equal(resp.json(), repo2_expected)
'Storage': '',
'Suite': ''}
all_repos = self.get("/api/publish")
self.check_equal(all_repos.status_code, 200)
self.check_in(repo_expected, all_repos.json())
self.check_exists("public/dists/" + distribution + "/Release")
self.check_exists("public/dists/" + distribution + "/main/binary-i386/Packages")
self.check_exists("public/dists/" + distribution + "/main/Contents-i386.gz")
self.check_exists("public/dists/" + distribution + "/main/binary-amd64/Packages")
self.check_not_exists("public/dists/" + distribution + "/main/Contents-amd64.gz")
self.check_exists("public/pool/main/b/boost-defaults/libboost-program-options-dev_1.49.0.1_i386.deb")
self.check_exists("public/dists/" + distribution +
"/main/binary-i386/Packages")
self.check_exists("public/dists/" + distribution +
"/main/Contents-i386.gz")
self.check_exists("public/dists/" + distribution +
"/main/binary-amd64/Packages")
self.check_not_exists(
"public/dists/" + distribution + "/main/Contents-amd64.gz")
self.check_exists(
"public/pool/main/b/boost-defaults/libboost-program-options-dev_1.49.0.1_i386.deb")
all_repos = self.get("/api/publish")
self.check_equal(all_repos.status_code, 200)
@@ -109,47 +128,68 @@ class PublishSnapshotAPITest(APITest):
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)
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)
"libboost-program-options-dev_1.49.0.1_i386.deb").status_code, 200)
self.check_equal(self.post("/api/repos/" + repo_name + "/file/" + d).status_code, 200)
self.check_equal(self.post_task("/api/repos/" + repo_name + "/file/" + d).json()['State'], 2)
self.check_equal(self.post("/api/repos/" + repo_name + '/snapshots', json={'Name': snapshot_name}).status_code, 201)
self.check_equal(self.post_task("/api/repos/" + repo_name + '/snapshots', json={'Name': snapshot_name}).json()['State'], 2)
prefix = self.random_name()
resp = self.post("/api/publish/" + prefix,
json={
"AcquireByHash": True,
"SourceKind": "snapshot",
"Sources": [{"Name": snapshot_name}],
"Signing": DefaultSigningOptions,
"Distribution": "squeeze",
"NotAutomatic": "yes",
"ButAutomaticUpgrades": "yes",
})
self.check_equal(resp.status_code, 201)
self.check_equal(resp.json(), {
resp = self.post_task(
"/api/publish/" + prefix,
json={
"AcquireByHash": True,
"SourceKind": "snapshot",
"Sources": [{"Name": snapshot_name}],
"Signing": DefaultSigningOptions,
"Distribution": "squeeze",
"NotAutomatic": "yes",
"ButAutomaticUpgrades": "yes",
"Origin": "earth",
"Label": "fun",
}
)
self.check_equal(resp.json()['State'], 2)
_id = resp.json()['ID']
resp = self.get("/api/tasks/" + str(_id) + "/detail")
self.check_equal(resp.json()['RemainingNumberOfPackages'], 0)
self.check_equal(resp.json()['TotalNumberOfPackages'], 1)
repo_expected = {
'AcquireByHash': True,
'Architectures': ['i386'],
'Distribution': 'squeeze',
'Label': '',
'Origin': '',
'Label': 'fun',
'Origin': 'earth',
'NotAutomatic': 'yes',
'ButAutomaticUpgrades': 'yes',
'Path': prefix + '/' + 'squeeze',
'Prefix': prefix,
'SkipContents': False,
'SourceKind': 'snapshot',
'Sources': [{'Component': 'main', 'Name': snapshot_name}],
'Storage': ''})
'Storage': '',
'Suite': '',
}
all_repos = self.get("/api/publish")
self.check_equal(all_repos.status_code, 200)
self.check_in(repo_expected, all_repos.json())
self.check_exists("public/" + prefix + "/dists/squeeze/Release")
self.check_exists("public/" + prefix + "/dists/squeeze/main/binary-i386/by-hash")
self.check_exists("public/" + prefix + "/dists/squeeze/main/binary-i386/Packages")
self.check_exists("public/" + prefix + "/dists/squeeze/main/Contents-i386.gz")
self.check_exists("public/" + prefix + "/pool/main/b/boost-defaults/libboost-program-options-dev_1.49.0.1_i386.deb")
self.check_exists("public/" + prefix +
"/dists/squeeze/main/binary-i386/by-hash")
self.check_exists("public/" + prefix +
"/dists/squeeze/main/binary-i386/Packages")
self.check_exists("public/" + prefix +
"/dists/squeeze/main/Contents-i386.gz")
self.check_exists(
"public/" + prefix + "/pool/main/b/boost-defaults/libboost-program-options-dev_1.49.0.1_i386.deb")
class PublishUpdateAPITestRepo(APITest):
@@ -160,7 +200,8 @@ class PublishUpdateAPITestRepo(APITest):
def check(self):
repo_name = self.random_name()
self.check_equal(self.post("/api/repos", json={"Name": repo_name, "DefaultDistribution": "wheezy"}).status_code, 201)
self.check_equal(self.post(
"/api/repos", json={"Name": repo_name, "DefaultDistribution": "wheezy"}).status_code, 201)
d = self.random_name()
self.check_equal(
@@ -168,36 +209,43 @@ class PublishUpdateAPITestRepo(APITest):
"pyspi_0.6.1-1.3.dsc",
"pyspi_0.6.1-1.3.diff.gz", "pyspi_0.6.1.orig.tar.gz",
"pyspi-0.6.1-1.3.stripped.dsc").status_code, 200)
self.check_equal(self.post("/api/repos/" + repo_name + "/file/" + d).status_code, 200)
self.check_equal(self.post_task("/api/repos/" + repo_name + "/file/" + d).json()['State'], 2)
prefix = self.random_name()
resp = self.post("/api/publish/" + prefix,
json={
"Architectures": ["i386", "source"],
"SourceKind": "local",
"Sources": [{"Name": repo_name}],
"Signing": DefaultSigningOptions,
})
resp = self.post_task(
"/api/publish/" + prefix,
json={
"Architectures": ["i386", "source"],
"SourceKind": "local",
"Sources": [{"Name": repo_name}],
"Signing": DefaultSigningOptions,
}
)
self.check_equal(resp.status_code, 201)
self.check_equal(resp.json()['State'], 2)
self.check_not_exists("public/" + prefix + "/pool/main/b/boost-defaults/libboost-program-options-dev_1.49.0.1_i386.deb")
self.check_exists("public/" + prefix + "/pool/main/p/pyspi/pyspi-0.6.1-1.3.stripped.dsc")
self.check_not_exists(
"public/" + prefix + "/pool/main/b/boost-defaults/libboost-program-options-dev_1.49.0.1_i386.deb")
self.check_exists("public/" + prefix +
"/pool/main/p/pyspi/pyspi-0.6.1-1.3.stripped.dsc")
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)
self.check_equal(self.post("/api/repos/" + repo_name + "/file/" + d).status_code, 200)
self.check_equal(self.post_task("/api/repos/" + repo_name + "/file/" + d).json()['State'], 2)
self.check_equal(self.delete("/api/repos/" + repo_name + "/packages/",
json={"PackageRefs": ['Psource pyspi 0.6.1-1.4 f8f1daa806004e89']}).status_code, 200)
self.check_equal(self.delete_task("/api/repos/" + repo_name + "/packages/",
json={"PackageRefs": ['Psource pyspi 0.6.1-1.4 f8f1daa806004e89']}).json()['State'], 2)
# Update and switch AcquireByHash on.
resp = self.put("/api/publish/" + prefix + "/wheezy",
json={
"AcquireByHash": True,
"Signing": DefaultSigningOptions,
})
resp = self.put_task(
"/api/publish/" + prefix + "/wheezy",
json={
"AcquireByHash": True,
"Signing": DefaultSigningOptions,
}
)
self.check_equal(resp.json()['State'], 2)
repo_expected = {
'AcquireByHash': True,
'Architectures': ['i386', 'source'],
@@ -206,21 +254,27 @@ class PublishUpdateAPITestRepo(APITest):
'Origin': '',
'NotAutomatic': '',
'ButAutomaticUpgrades': '',
'Path': prefix + '/' + 'wheezy',
'Prefix': prefix,
'SkipContents': False,
'SourceKind': 'local',
'Sources': [{'Component': 'main', 'Name': repo_name}],
'Storage': ''}
'Storage': '',
'Suite': ''}
self.check_equal(resp.status_code, 200)
self.check_equal(resp.json(), repo_expected)
all_repos = self.get("/api/publish")
self.check_equal(all_repos.status_code, 200)
self.check_in(repo_expected, all_repos.json())
self.check_exists("public/" + prefix + "/dists/wheezy/main/binary-i386/by-hash")
self.check_exists("public/" + prefix +
"/dists/wheezy/main/binary-i386/by-hash")
self.check_exists("public/" + prefix + "/pool/main/b/boost-defaults/libboost-program-options-dev_1.49.0.1_i386.deb")
self.check_not_exists("public/" + prefix + "/pool/main/p/pyspi/pyspi-0.6.1-1.3.stripped.dsc")
self.check_exists(
"public/" + prefix + "/pool/main/b/boost-defaults/libboost-program-options-dev_1.49.0.1_i386.deb")
self.check_not_exists(
"public/" + prefix + "/pool/main/p/pyspi/pyspi-0.6.1-1.3.stripped.dsc")
self.check_equal(self.delete("/api/publish/" + prefix + "/wheezy").status_code, 200)
self.check_equal(self.delete_task("/api/publish/" + prefix + "/wheezy").json()['State'], 2)
self.check_not_exists("public/" + prefix + "dists/")
@@ -232,7 +286,8 @@ class PublishUpdateSkipCleanupAPITestRepo(APITest):
def check(self):
repo_name = self.random_name()
self.check_equal(self.post("/api/repos", json={"Name": repo_name, "DefaultDistribution": "wheezy"}).status_code, 201)
self.check_equal(self.post(
"/api/repos", json={"Name": repo_name, "DefaultDistribution": "wheezy"}).status_code, 201)
d = self.random_name()
self.check_equal(
@@ -240,48 +295,51 @@ class PublishUpdateSkipCleanupAPITestRepo(APITest):
"pyspi_0.6.1-1.3.dsc",
"pyspi_0.6.1-1.3.diff.gz", "pyspi_0.6.1.orig.tar.gz",
"pyspi-0.6.1-1.3.stripped.dsc").status_code, 200)
self.check_equal(self.post("/api/repos/" + repo_name + "/file/" + d).status_code, 200)
self.check_equal(self.post_task("/api/repos/" + repo_name + "/file/" + d).json()['State'], 2)
prefix = self.random_name()
resp = self.post("/api/publish/" + prefix,
json={
"Architectures": ["i386", "source"],
"SourceKind": "local",
"Sources": [{"Name": repo_name}],
"Signing": DefaultSigningOptions,
})
resp = self.post_task("/api/publish/" + prefix,
json={
"Architectures": ["i386", "source"],
"SourceKind": "local",
"Sources": [{"Name": repo_name}],
"Signing": DefaultSigningOptions,
})
self.check_equal(resp.status_code, 201)
self.check_equal(resp.json()['State'], 2)
self.check_not_exists("public/" + prefix + "/pool/main/b/boost-defaults/libboost-program-options-dev_1.49.0.1_i386.deb")
self.check_exists("public/" + prefix + "/pool/main/p/pyspi/pyspi-0.6.1-1.3.stripped.dsc")
self.check_not_exists(
"public/" + prefix + "/pool/main/b/boost-defaults/libboost-program-options-dev_1.49.0.1_i386.deb")
self.check_exists("public/" + prefix +
"/pool/main/p/pyspi/pyspi-0.6.1-1.3.stripped.dsc")
# Publish two repos, so that deleting one while skipping cleanup will
# not delete the whole prefix.
resp = self.post("/api/publish/" + prefix,
json={
"Architectures": ["i386", "source"],
"Distribution": "otherdist",
"SourceKind": "local",
"Sources": [{"Name": repo_name}],
"Signing": DefaultSigningOptions,
})
resp = self.post_task("/api/publish/" + prefix,
json={
"Architectures": ["i386", "source"],
"Distribution": "otherdist",
"SourceKind": "local",
"Sources": [{"Name": repo_name}],
"Signing": DefaultSigningOptions,
})
self.check_equal(resp.status_code, 201)
self.check_equal(resp.json()['State'], 2)
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)
self.check_equal(self.post("/api/repos/" + repo_name + "/file/" + d).status_code, 200)
self.check_equal(self.post_task("/api/repos/" + repo_name + "/file/" + d).json()['State'], 2)
self.check_equal(self.delete("/api/repos/" + repo_name + "/packages/",
json={"PackageRefs": ['Psource pyspi 0.6.1-1.4 f8f1daa806004e89']}).status_code, 200)
self.check_equal(self.delete_task("/api/repos/" + repo_name + "/packages/",
json={"PackageRefs": ['Psource pyspi 0.6.1-1.4 f8f1daa806004e89']}).json()['State'], 2)
resp = self.put("/api/publish/" + prefix + "/wheezy",
json={
"Signing": DefaultSigningOptions,
"SkipCleanup": True,
})
resp = self.put_task("/api/publish/" + prefix + "/wheezy",
json={
"Signing": DefaultSigningOptions,
"SkipCleanup": True,
})
self.check_equal(resp.json()['State'], 2)
repo_expected = {
'AcquireByHash': False,
'Architectures': ['i386', 'source'],
@@ -290,19 +348,24 @@ class PublishUpdateSkipCleanupAPITestRepo(APITest):
'Origin': '',
'NotAutomatic': '',
'ButAutomaticUpgrades': '',
'Path': prefix + '/' + 'wheezy',
'Prefix': prefix,
'SkipContents': False,
'SourceKind': 'local',
'Sources': [{'Component': 'main', 'Name': repo_name}],
'Storage': ''}
'Storage': '',
'Suite': ''}
self.check_equal(resp.status_code, 200)
self.check_equal(resp.json(), repo_expected)
all_repos = self.get("/api/publish")
self.check_equal(all_repos.status_code, 200)
self.check_in(repo_expected, all_repos.json())
self.check_exists("public/" + prefix + "/pool/main/b/boost-defaults/libboost-program-options-dev_1.49.0.1_i386.deb")
self.check_exists("public/" + prefix + "/pool/main/p/pyspi/pyspi-0.6.1-1.3.stripped.dsc")
self.check_exists(
"public/" + prefix + "/pool/main/b/boost-defaults/libboost-program-options-dev_1.49.0.1_i386.deb")
self.check_exists("public/" + prefix +
"/pool/main/p/pyspi/pyspi-0.6.1-1.3.stripped.dsc")
self.check_equal(self.delete("/api/publish/" + prefix + "/wheezy", params={"SkipCleanup": "1"}).status_code, 200)
self.check_equal(self.delete_task("/api/publish/" + prefix + "/wheezy", params={"SkipCleanup": "1"}).json()['State'], 2)
self.check_exists("public/" + prefix + "/pool/main/b/boost-defaults/libboost-program-options-dev_1.49.0.1_i386.deb")
self.check_exists("public/" + prefix + "/pool/main/p/pyspi/pyspi-0.6.1-1.3.stripped.dsc")
@@ -315,7 +378,8 @@ class PublishSwitchAPITestRepo(APITest):
def check(self):
repo_name = self.random_name()
self.check_equal(self.post("/api/repos", json={"Name": repo_name, "DefaultDistribution": "wheezy"}).status_code, 201)
self.check_equal(self.post(
"/api/repos", json={"Name": repo_name, "DefaultDistribution": "wheezy"}).status_code, 201)
d = self.random_name()
self.check_equal(
@@ -323,21 +387,22 @@ class PublishSwitchAPITestRepo(APITest):
"pyspi_0.6.1-1.3.dsc",
"pyspi_0.6.1-1.3.diff.gz", "pyspi_0.6.1.orig.tar.gz",
"pyspi-0.6.1-1.3.stripped.dsc").status_code, 200)
self.check_equal(self.post("/api/repos/" + repo_name + "/file/" + d).status_code, 200)
self.check_equal(self.post_task("/api/repos/" + repo_name + "/file/" + d).json()['State'], 2)
snapshot1_name = self.random_name()
self.check_equal(self.post("/api/repos/" + repo_name + '/snapshots', json={'Name': snapshot1_name}).status_code, 201)
self.check_equal(self.post_task("/api/repos/" + repo_name + '/snapshots', json={'Name': snapshot1_name}).json()['State'], 2)
prefix = self.random_name()
resp = self.post("/api/publish/" + prefix,
json={
"Architectures": ["i386", "source"],
"SourceKind": "snapshot",
"Sources": [{"Name": snapshot1_name}],
"Signing": DefaultSigningOptions,
})
resp = self.post_task(
"/api/publish/" + prefix,
json={
"Architectures": ["i386", "source"],
"SourceKind": "snapshot",
"Sources": [{"Name": snapshot1_name}],
"Signing": DefaultSigningOptions,
})
self.check_equal(resp.status_code, 201)
self.check_equal(resp.json()['State'], 2)
repo_expected = {
'AcquireByHash': False,
'Architectures': ['i386', 'source'],
@@ -346,33 +411,41 @@ class PublishSwitchAPITestRepo(APITest):
'NotAutomatic': '',
'ButAutomaticUpgrades': '',
'Origin': '',
'Path': prefix + '/' + 'wheezy',
'Prefix': prefix,
'SkipContents': False,
'SourceKind': 'snapshot',
'Sources': [{'Component': 'main', 'Name': snapshot1_name}],
'Storage': ''}
self.check_equal(resp.json(), repo_expected)
'Storage': '',
'Suite': ''}
all_repos = self.get("/api/publish")
self.check_equal(all_repos.status_code, 200)
self.check_in(repo_expected, all_repos.json())
self.check_not_exists("public/" + prefix + "/pool/main/b/boost-defaults/libboost-program-options-dev_1.49.0.1_i386.deb")
self.check_exists("public/" + prefix + "/pool/main/p/pyspi/pyspi-0.6.1-1.3.stripped.dsc")
self.check_not_exists(
"public/" + prefix + "/pool/main/b/boost-defaults/libboost-program-options-dev_1.49.0.1_i386.deb")
self.check_exists("public/" + prefix +
"/pool/main/p/pyspi/pyspi-0.6.1-1.3.stripped.dsc")
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)
self.check_equal(self.post("/api/repos/" + repo_name + "/file/" + d).status_code, 200)
self.check_equal(self.post_task("/api/repos/" + repo_name + "/file/" + d).json()['State'], 2)
self.check_equal(self.delete("/api/repos/" + repo_name + "/packages/",
json={"PackageRefs": ['Psource pyspi 0.6.1-1.4 f8f1daa806004e89']}).status_code, 200)
self.check_equal(self.delete_task("/api/repos/" + repo_name + "/packages/",
json={"PackageRefs": ['Psource pyspi 0.6.1-1.4 f8f1daa806004e89']}).json()['State'], 2)
snapshot2_name = self.random_name()
self.check_equal(self.post("/api/repos/" + repo_name + '/snapshots', json={'Name': snapshot2_name}).status_code, 201)
self.check_equal(self.post_task("/api/repos/" + repo_name + '/snapshots', json={'Name': snapshot2_name}).json()['State'], 2)
resp = self.put("/api/publish/" + prefix + "/wheezy",
json={
"Snapshots": [{"Component": "main", "Name": snapshot2_name}],
"Signing": DefaultSigningOptions,
"SkipContents": True,
})
resp = self.put_task(
"/api/publish/" + prefix + "/wheezy",
json={
"Snapshots": [{"Component": "main", "Name": snapshot2_name}],
"Signing": DefaultSigningOptions,
"SkipContents": True,
})
self.check_equal(resp.json()['State'], 2)
repo_expected = {
'AcquireByHash': False,
'Architectures': ['i386', 'source'],
@@ -381,19 +454,24 @@ class PublishSwitchAPITestRepo(APITest):
'Origin': '',
'NotAutomatic': '',
'ButAutomaticUpgrades': '',
'Path': prefix + '/' + 'wheezy',
'Prefix': prefix,
'SkipContents': True,
'SourceKind': 'snapshot',
'Sources': [{'Component': 'main', 'Name': snapshot2_name}],
'Storage': ''}
'Storage': '',
'Suite': ''}
self.check_equal(resp.status_code, 200)
self.check_equal(resp.json(), repo_expected)
all_repos = self.get("/api/publish")
self.check_equal(all_repos.status_code, 200)
self.check_in(repo_expected, all_repos.json())
self.check_exists("public/" + prefix + "/pool/main/b/boost-defaults/libboost-program-options-dev_1.49.0.1_i386.deb")
self.check_not_exists("public/" + prefix + "/pool/main/p/pyspi/pyspi-0.6.1-1.3.stripped.dsc")
self.check_exists(
"public/" + prefix + "/pool/main/b/boost-defaults/libboost-program-options-dev_1.49.0.1_i386.deb")
self.check_not_exists(
"public/" + prefix + "/pool/main/p/pyspi/pyspi-0.6.1-1.3.stripped.dsc")
self.check_equal(self.delete("/api/publish/" + prefix + "/wheezy").status_code, 200)
self.check_equal(self.delete_task("/api/publish/" + prefix + "/wheezy").json()['State'], 2)
self.check_not_exists("public/" + prefix + "dists/")
@@ -405,7 +483,8 @@ class PublishSwitchAPISkipCleanupTestRepo(APITest):
def check(self):
repo_name = self.random_name()
self.check_equal(self.post("/api/repos", json={"Name": repo_name, "DefaultDistribution": "wheezy"}).status_code, 201)
self.check_equal(self.post(
"/api/repos", json={"Name": repo_name, "DefaultDistribution": "wheezy"}).status_code, 201)
d = self.random_name()
self.check_equal(
@@ -413,21 +492,21 @@ class PublishSwitchAPISkipCleanupTestRepo(APITest):
"pyspi_0.6.1-1.3.dsc",
"pyspi_0.6.1-1.3.diff.gz", "pyspi_0.6.1.orig.tar.gz",
"pyspi-0.6.1-1.3.stripped.dsc").status_code, 200)
self.check_equal(self.post("/api/repos/" + repo_name + "/file/" + d).status_code, 200)
self.check_equal(self.post_task("/api/repos/" + repo_name + "/file/" + d).json()['State'], 2)
snapshot1_name = self.random_name()
self.check_equal(self.post("/api/repos/" + repo_name + '/snapshots', json={'Name': snapshot1_name}).status_code, 201)
self.check_equal(self.post_task("/api/repos/" + repo_name + '/snapshots', json={'Name': snapshot1_name}).json()['State'], 2)
prefix = self.random_name()
resp = self.post("/api/publish/" + prefix,
json={
"Architectures": ["i386", "source"],
"SourceKind": "snapshot",
"Sources": [{"Name": snapshot1_name}],
"Signing": DefaultSigningOptions,
})
resp = self.post_task("/api/publish/" + prefix,
json={
"Architectures": ["i386", "source"],
"SourceKind": "snapshot",
"Sources": [{"Name": snapshot1_name}],
"Signing": DefaultSigningOptions,
})
self.check_equal(resp.status_code, 201)
self.check_equal(resp.json()['State'], 2)
repo_expected = {
'AcquireByHash': False,
'Architectures': ['i386', 'source'],
@@ -436,28 +515,34 @@ class PublishSwitchAPISkipCleanupTestRepo(APITest):
'NotAutomatic': '',
'ButAutomaticUpgrades': '',
'Origin': '',
'Path': prefix + '/' + 'wheezy',
'Prefix': prefix,
'SkipContents': False,
'SourceKind': 'snapshot',
'Sources': [{'Component': 'main', 'Name': snapshot1_name}],
'Storage': ''}
self.check_equal(resp.json(), repo_expected)
'Storage': '',
'Suite': ''}
all_repos = self.get("/api/publish")
self.check_equal(all_repos.status_code, 200)
self.check_in(repo_expected, all_repos.json())
self.check_not_exists("public/" + prefix + "/pool/main/b/boost-defaults/libboost-program-options-dev_1.49.0.1_i386.deb")
self.check_exists("public/" + prefix + "/pool/main/p/pyspi/pyspi-0.6.1-1.3.stripped.dsc")
self.check_not_exists(
"public/" + prefix + "/pool/main/b/boost-defaults/libboost-program-options-dev_1.49.0.1_i386.deb")
self.check_exists("public/" + prefix +
"/pool/main/p/pyspi/pyspi-0.6.1-1.3.stripped.dsc")
# Publish two snapshots, so that deleting one while skipping cleanup will
# not delete the whole prefix.
resp = self.post("/api/publish/" + prefix,
json={
"Architectures": ["i386", "source"],
"Distribution": "otherdist",
"SourceKind": "snapshot",
"Sources": [{"Name": snapshot1_name}],
"Signing": DefaultSigningOptions,
})
resp = self.post_task("/api/publish/" + prefix,
json={
"Architectures": ["i386", "source"],
"Distribution": "otherdist",
"SourceKind": "snapshot",
"Sources": [{"Name": snapshot1_name}],
"Signing": DefaultSigningOptions,
})
self.check_equal(resp.status_code, 201)
self.check_equal(resp.json()['State'], 2)
repo_expected = {
'AcquireByHash': False,
'Architectures': ['i386', 'source'],
@@ -466,31 +551,36 @@ class PublishSwitchAPISkipCleanupTestRepo(APITest):
'NotAutomatic': '',
'ButAutomaticUpgrades': '',
'Origin': '',
'Path': prefix + '/' + 'otherdist',
'Prefix': prefix,
'SkipContents': False,
'SourceKind': 'snapshot',
'Sources': [{'Component': 'main', 'Name': snapshot1_name}],
'Storage': ''}
self.check_equal(resp.json(), repo_expected)
'Storage': '',
'Suite': ''}
all_repos = self.get("/api/publish")
self.check_equal(all_repos.status_code, 200)
self.check_in(repo_expected, all_repos.json())
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)
self.check_equal(self.post("/api/repos/" + repo_name + "/file/" + d).status_code, 200)
self.check_equal(self.post_task("/api/repos/" + repo_name + "/file/" + d).json()['State'], 2)
self.check_equal(self.delete("/api/repos/" + repo_name + "/packages/",
json={"PackageRefs": ['Psource pyspi 0.6.1-1.4 f8f1daa806004e89']}).status_code, 200)
self.check_equal(self.delete_task("/api/repos/" + repo_name + "/packages/",
json={"PackageRefs": ['Psource pyspi 0.6.1-1.4 f8f1daa806004e89']}).json()['State'], 2)
snapshot2_name = self.random_name()
self.check_equal(self.post("/api/repos/" + repo_name + '/snapshots', json={'Name': snapshot2_name}).status_code, 201)
self.check_equal(self.post_task("/api/repos/" + repo_name + '/snapshots', json={'Name': snapshot2_name}).json()['State'], 2)
resp = self.put("/api/publish/" + prefix + "/wheezy",
json={
"Snapshots": [{"Component": "main", "Name": snapshot2_name}],
"Signing": DefaultSigningOptions,
"SkipCleanup": True,
"SkipContents": True,
})
resp = self.put_task("/api/publish/" + prefix + "/wheezy",
json={
"Snapshots": [{"Component": "main", "Name": snapshot2_name}],
"Signing": DefaultSigningOptions,
"SkipCleanup": True,
"SkipContents": True,
})
self.check_equal(resp.json()['State'], 2)
repo_expected = {
'AcquireByHash': False,
'Architectures': ['i386', 'source'],
@@ -499,18 +589,21 @@ class PublishSwitchAPISkipCleanupTestRepo(APITest):
'Origin': '',
'NotAutomatic': '',
'ButAutomaticUpgrades': '',
'Path': prefix + '/' + 'wheezy',
'Prefix': prefix,
'SkipContents': True,
'SourceKind': 'snapshot',
'Sources': [{'Component': 'main', 'Name': snapshot2_name}],
'Storage': ''}
'Storage': '',
'Suite': ''}
self.check_equal(resp.status_code, 200)
self.check_equal(resp.json(), repo_expected)
all_repos = self.get("/api/publish")
self.check_equal(all_repos.status_code, 200)
self.check_in(repo_expected, all_repos.json())
self.check_exists("public/" + prefix + "/pool/main/b/boost-defaults/libboost-program-options-dev_1.49.0.1_i386.deb")
self.check_exists("public/" + prefix + "/pool/main/p/pyspi/pyspi-0.6.1-1.3.stripped.dsc")
self.check_equal(self.delete("/api/publish/" + prefix + "/wheezy", params={"SkipCleanup": "1"}).status_code, 200)
self.check_equal(self.delete_task("/api/publish/" + prefix + "/wheezy", params={"SkipCleanup": "1"}).json()['State'], 2)
self.check_exists("public/" + prefix + "/pool/main/b/boost-defaults/libboost-program-options-dev_1.49.0.1_i386.deb")
self.check_exists("public/" + prefix + "/pool/main/p/pyspi/pyspi-0.6.1-1.3.stripped.dsc")
system/t12_api/repos.py
+58 -57
View File
@@ -1,5 +1,5 @@
from api_lib import APITest
from publish import DefaultSigningOptions
from .publish import DefaultSigningOptions
class ReposAPITestCreateShow(APITest):
@@ -8,10 +8,10 @@ class ReposAPITestCreateShow(APITest):
"""
def check(self):
repo_name = self.random_name()
repo_desc = {u'Comment': u'fun repo',
u'DefaultComponent': u'',
u'DefaultDistribution': u'',
u'Name': repo_name}
repo_desc = {'Comment': 'fun repo',
'DefaultComponent': '',
'DefaultDistribution': '',
'Name': repo_name}
resp = self.post("/api/repos", json={"Name": repo_name, "Comment": "fun repo"})
self.check_equal(resp.json(), repo_desc)
@@ -40,7 +40,7 @@ class ReposAPITestCreateIndexDelete(APITest):
names = [repo["Name"] for repo in repos]
assert repo_name in names
self.check_equal(self.delete("/api/repos/" + repo_name).status_code, 200)
self.check_equal(self.delete_task("/api/repos/" + repo_name).json()['State'], 2)
self.check_equal(self.delete("/api/repos/" + repo_name).status_code, 404)
self.check_equal(self.get("/api/repos/" + repo_name).status_code, 404)
@@ -59,26 +59,26 @@ class ReposAPITestCreateIndexDelete(APITest):
self.check_equal(self.upload("/api/files/" + d,
"pyspi_0.6.1-1.3.dsc", "pyspi_0.6.1-1.3.diff.gz", "pyspi_0.6.1.orig.tar.gz").status_code, 200)
resp = self.post("/api/repos/" + repo_name + "/file/" + d)
self.check_equal(resp.status_code, 200)
resp = self.post_task("/api/repos/" + repo_name + "/file/" + d)
self.check_equal(resp.json()['State'], 2)
self.check_equal(self.post("/api/repos/" + repo_name + "/snapshots", json={"Name": repo_name}).status_code, 201)
self.check_equal(self.post_task("/api/repos/" + repo_name + "/snapshots", json={"Name": repo_name}).json()['State'], 2)
self.check_equal(self.post("/api/publish",
self.check_equal(self.post_task("/api/publish",
json={
"SourceKind": "local",
"Sources": [{"Name": repo_name}],
"Signing": DefaultSigningOptions,
}).status_code, 201)
}).json()['State'], 2)
# repo is not deletable while it is published
self.check_equal(self.delete("/api/repos/" + repo_name).status_code, 409)
self.check_equal(self.delete("/api/repos/" + repo_name, params={"force": "1"}).status_code, 409)
self.check_equal(self.delete_task("/api/repos/" + repo_name).json()['State'], 3)
self.check_equal(self.delete_task("/api/repos/" + repo_name, params={"force": "1"}).json()['State'], 3)
# drop published
self.check_equal(self.delete("/api/publish//" + distribution).status_code, 200)
self.check_equal(self.delete("/api/repos/" + repo_name).status_code, 409)
self.check_equal(self.delete("/api/repos/" + repo_name, params={"force": "1"}).status_code, 200)
self.check_equal(self.delete_task("/api/publish//" + distribution).json()['State'], 2)
self.check_equal(self.delete_task("/api/repos/" + repo_name).json()['State'], 3)
self.check_equal(self.delete_task("/api/repos/" + repo_name, params={"force": "1"}).json()['State'], 2)
self.check_equal(self.get("/api/repos/" + repo_name).status_code, 404)
@@ -95,14 +95,16 @@ class ReposAPITestAdd(APITest):
self.check_equal(self.upload("/api/files/" + d,
"pyspi_0.6.1-1.3.dsc", "pyspi_0.6.1-1.3.diff.gz", "pyspi_0.6.1.orig.tar.gz").status_code, 200)
resp = self.post("/api/repos/" + repo_name + "/file/" + d)
resp = self.post_task("/api/repos/" + repo_name + "/file/" + d)
self.check_equal(resp.json()['State'], 2)
resp = self.get("/api/tasks/" + str(resp.json()['ID']) + "/output")
self.check_equal(resp.status_code, 200)
self.check_equal(resp.json(), {
u'FailedFiles': [],
u'Report': {
u'Added': [u'pyspi_0.6.1-1.3_source added'],
u'Removed': [],
u'Warnings': []}})
self.check_in(b"Added: pyspi_0.6.1-1.3_source added", resp.content)
self.check_not_in(b"Removed: ", resp.content)
self.check_not_in(b"Failed files: ", resp.content)
self.check_not_in(b"Warnings: ", resp.content)
self.check_equal(self.get("/api/repos/" + repo_name + "/packages").json(), ['Psource pyspi 0.6.1-1.3 3a8b37cbd9a3559e'])
@@ -122,7 +124,7 @@ class ReposAPITestAddNotFullRemove(APITest):
self.check_equal(self.upload("/api/files/" + d,
"pyspi_0.6.1-1.3.dsc", "pyspi_0.6.1-1.3.diff.gz", "pyspi_0.6.1.orig.tar.gz", "aptly.pub").status_code, 200)
self.check_equal(self.post("/api/repos/" + repo_name + "/file/" + d).status_code, 200)
self.check_equal(self.post_task("/api/repos/" + repo_name + "/file/" + d).json()['State'], 2)
self.check_equal(self.get("/api/repos/" + repo_name + "/packages").json(), ['Psource pyspi 0.6.1-1.3 3a8b37cbd9a3559e'])
self.check_exists("upload/" + d + "/aptly.pub")
@@ -142,7 +144,7 @@ class ReposAPITestAddNoRemove(APITest):
self.check_equal(self.upload("/api/files/" + d,
"pyspi_0.6.1-1.3.dsc", "pyspi_0.6.1-1.3.diff.gz", "pyspi_0.6.1.orig.tar.gz").status_code, 200)
self.check_equal(self.post("/api/repos/" + repo_name + "/file/" + d, params={"noRemove": 1}).status_code, 200)
self.check_equal(self.post_task("/api/repos/" + repo_name + "/file/" + d, params={"noRemove": 1}).json()['State'], 2)
self.check_equal(self.get("/api/repos/" + repo_name + "/packages").json(), ['Psource pyspi 0.6.1-1.3 3a8b37cbd9a3559e'])
self.check_exists("upload/" + d + "/pyspi_0.6.1-1.3.dsc")
@@ -161,14 +163,16 @@ class ReposAPITestAddFile(APITest):
self.check_equal(self.upload("/api/files/" + d,
"libboost-program-options-dev_1.49.0.1_i386.deb").status_code, 200)
resp = self.post("/api/repos/" + repo_name + "/file/" + d + "/libboost-program-options-dev_1.49.0.1_i386.deb")
resp = self.post_task("/api/repos/" + repo_name + "/file/" + d + "/libboost-program-options-dev_1.49.0.1_i386.deb")
self.check_equal(resp.json()['State'], 2)
resp = self.get("/api/tasks/" + str(resp.json()['ID']) + "/output")
self.check_equal(resp.status_code, 200)
self.check_equal(resp.json(), {
u'FailedFiles': [],
u'Report': {
u'Added': [u'libboost-program-options-dev_1.49.0.1_i386 added'],
u'Removed': [],
u'Warnings': []}})
self.check_in(b"Added: libboost-program-options-dev_1.49.0.1_i386 added", resp.content)
self.check_not_in(b"Removed: ", resp.content)
self.check_not_in(b"Failed files: ", resp.content)
self.check_not_in(b"Warnings: ", resp.content)
self.check_equal(self.get("/api/repos/" + repo_name + "/packages").json(),
['Pi386 libboost-program-options-dev 1.49.0.1 918d2f433384e378'])
@@ -191,18 +195,15 @@ class ReposAPITestInclude(APITest):
"hardlink_0.2.1_amd64.deb", directory='changes')
self.check_equal(resp.status_code, 200)
resp = self.post("/api/repos/" + repo_name + "/include/" + d, params={"ignoreSignature": 1})
self.check_equal(resp.status_code, 200)
self.check_equal(resp.json(), {
u'FailedFiles': [],
u'Report': {
u'Added': [u'hardlink_0.2.1_source added', 'hardlink_0.2.1_amd64 added'],
u'Removed': [],
u'Warnings': []}})
resp = self.post_task("/api/repos/" + repo_name + "/include/" + d, params={"ignoreSignature": 1})
self.check_equal(resp.json()['State'], 2)
resp = self.get("/api/tasks/" + str(resp.json()['ID']) + "/output")
self.check_equal(resp.status_code, 200)
self.check_in(b"Added: hardlink_0.2.1_source added, hardlink_0.2.1_amd64 added", resp.content)
self.check_equal(
sorted(self.get("/api/repos/" + repo_name + "/packages").json()),
[u'Pamd64 hardlink 0.2.1 daf8fcecbf8210ad', u'Psource hardlink 0.2.1 8f72df429d7166e5']
['Pamd64 hardlink 0.2.1 daf8fcecbf8210ad', 'Psource hardlink 0.2.1 8f72df429d7166e5']
)
self.check_not_exists("upload/" + d)
@@ -223,7 +224,7 @@ class ReposAPITestShowQuery(APITest):
"libboost-program-options-dev_1.49.0.1_i386.deb", "pyspi_0.6.1-1.3.dsc",
"pyspi_0.6.1-1.3.diff.gz", "pyspi_0.6.1.orig.tar.gz",
"pyspi-0.6.1-1.3.stripped.dsc").status_code, 200)
self.check_equal(self.post("/api/repos/" + repo_name + "/file/" + d).status_code, 200)
self.check_equal(self.post_task("/api/repos/" + repo_name + "/file/" + d).json()['State'], 2)
self.check_equal(sorted(self.get("/api/repos/" + repo_name + "/packages", params={"q": "pyspi"}).json()),
['Psource pyspi 0.6.1-1.3 3a8b37cbd9a3559e', 'Psource pyspi 0.6.1-1.4 f8f1daa806004e89'])
@@ -237,7 +238,7 @@ class ReposAPITestShowQuery(APITest):
resp = self.get("/api/repos/" + repo_name + "/packages", params={"q": "pyspi)"})
self.check_equal(resp.status_code, 400)
self.check_equal(resp.json()["error"], u'parsing failed: unexpected token ): expecting end of query')
self.check_equal(resp.json()["error"], 'parsing failed: unexpected token ): expecting end of query')
class ReposAPITestAddMultiple(APITest):
@@ -256,13 +257,13 @@ class ReposAPITestAddMultiple(APITest):
"pyspi_0.6.1-1.3.diff.gz", "pyspi_0.6.1.orig.tar.gz",
"pyspi-0.6.1-1.3.stripped.dsc").status_code, 200)
self.check_equal(self.post("/api/repos/" + repo_name + "/file/" + d + "/pyspi_0.6.1-1.3.dsc",
params={"noRemove": 1}).status_code, 200)
self.check_equal(self.post_task("/api/repos/" + repo_name + "/file/" + d + "/pyspi_0.6.1-1.3.dsc",
params={"noRemove": 1}).json()['State'], 2)
self.check_equal(sorted(self.get("/api/repos/" + repo_name + "/packages").json()),
['Psource pyspi 0.6.1-1.3 3a8b37cbd9a3559e'])
self.check_equal(self.post("/api/repos/" + repo_name + "/file/" + d + "/pyspi-0.6.1-1.3.stripped.dsc").status_code, 200)
self.check_equal(self.post_task("/api/repos/" + repo_name + "/file/" + d + "/pyspi-0.6.1-1.3.stripped.dsc").json()['State'], 2)
self.check_equal(sorted(self.get("/api/repos/" + repo_name + "/packages").json()),
['Psource pyspi 0.6.1-1.3 3a8b37cbd9a3559e', 'Psource pyspi 0.6.1-1.4 f8f1daa806004e89'])
@@ -284,34 +285,34 @@ class ReposAPITestPackagesAddDelete(APITest):
"pyspi_0.6.1-1.3.diff.gz", "pyspi_0.6.1.orig.tar.gz",
"pyspi-0.6.1-1.3.stripped.dsc").status_code, 200)
self.check_equal(self.post("/api/repos/" + repo_name + "/file/" + d).status_code, 200)
self.check_equal(self.post_task("/api/repos/" + repo_name + "/file/" + d).json()['State'], 2)
self.check_equal(sorted(self.get("/api/repos/" + repo_name + "/packages").json()),
['Pi386 libboost-program-options-dev 1.49.0.1 918d2f433384e378',
'Psource pyspi 0.6.1-1.3 3a8b37cbd9a3559e',
'Psource pyspi 0.6.1-1.4 f8f1daa806004e89'])
self.check_equal(self.post("/api/repos/" + repo_name + "/packages/",
json={"PackageRefs": ['Psource pyspi 0.6.1-1.4 f8f1daa806004e89']}).status_code, 200)
self.check_equal(self.post_task("/api/repos/" + repo_name + "/packages/",
json={"PackageRefs": ['Psource pyspi 0.6.1-1.4 f8f1daa806004e89']}).json()['State'], 2)
self.check_equal(sorted(self.get("/api/repos/" + repo_name + "/packages").json()),
['Pi386 libboost-program-options-dev 1.49.0.1 918d2f433384e378',
'Psource pyspi 0.6.1-1.3 3a8b37cbd9a3559e',
'Psource pyspi 0.6.1-1.4 f8f1daa806004e89'])
self.check_equal(self.post("/api/repos/" + repo_name + "/packages/",
self.check_equal(self.post_task("/api/repos/" + repo_name + "/packages/",
json={"PackageRefs": ['Psource pyspi 0.6.1-1.4 f8f1daa806004e89',
'Psource no-such-package 0.6.1-1.4 f8f1daa806004e89']}).status_code, 404)
'Psource no-such-package 0.6.1-1.4 f8f1daa806004e89']}).json()['State'], 3)
self.check_equal(self.delete("/api/repos/" + repo_name + "/packages/",
json={"PackageRefs": ['Psource pyspi 0.6.1-1.4 f8f1daa806004e89']}).status_code, 200)
self.check_equal(self.delete_task("/api/repos/" + repo_name + "/packages/",
json={"PackageRefs": ['Psource pyspi 0.6.1-1.4 f8f1daa806004e89']}).json()['State'], 2)
self.check_equal(sorted(self.get("/api/repos/" + repo_name + "/packages").json()),
['Pi386 libboost-program-options-dev 1.49.0.1 918d2f433384e378',
'Psource pyspi 0.6.1-1.3 3a8b37cbd9a3559e'])
self.check_equal(self.post("/api/repos/" + repo_name + "/packages/",
json={"PackageRefs": ['Psource pyspi 0.6.1-1.4 f8f1daa806004e89']}).status_code, 200)
self.check_equal(self.post_task("/api/repos/" + repo_name + "/packages/",
json={"PackageRefs": ['Psource pyspi 0.6.1-1.4 f8f1daa806004e89']}).json()['State'], 2)
self.check_equal(sorted(self.get("/api/repos/" + repo_name + "/packages").json()),
['Pi386 libboost-program-options-dev 1.49.0.1 918d2f433384e378',
@@ -322,9 +323,9 @@ class ReposAPITestPackagesAddDelete(APITest):
self.check_equal(self.post("/api/repos", json={"Name": repo_name2, "Comment": "fun repo"}).status_code, 201)
self.check_equal(self.post("/api/repos/" + repo_name2 + "/packages/",
self.check_equal(self.post_task("/api/repos/" + repo_name2 + "/packages/",
json={"PackageRefs": ['Psource pyspi 0.6.1-1.4 f8f1daa806004e89',
'Pi386 libboost-program-options-dev 1.49.0.1 918d2f433384e378']}).status_code, 200)
'Pi386 libboost-program-options-dev 1.49.0.1 918d2f433384e378']}).json()['State'], 2)
self.check_equal(sorted(self.get("/api/repos/" + repo_name2 + "/packages").json()),
['Pi386 libboost-program-options-dev 1.49.0.1 918d2f433384e378',
system/t12_api/snapshots.py
+81 -72
View File
@@ -1,5 +1,5 @@
from api_lib import APITest
from publish import DefaultSigningOptions
from .publish import DefaultSigningOptions
class SnapshotsAPITestCreateShowEmpty(APITest):
@@ -8,13 +8,12 @@ class SnapshotsAPITestCreateShowEmpty(APITest):
"""
def check(self):
snapshot_name = self.random_name()
snapshot_desc = {u'Description': u'fun snapshot',
u'Name': snapshot_name}
snapshot_desc = {'Description': 'fun snapshot',
'Name': snapshot_name}
# create empty snapshot
resp = self.post("/api/snapshots", json=snapshot_desc)
self.check_subset(snapshot_desc, resp.json())
self.check_equal(resp.status_code, 201)
resp = self.post_task("/api/snapshots", json=snapshot_desc)
self.check_equal(resp.json()['State'], 2)
self.check_subset(snapshot_desc, self.get("/api/snapshots/" + snapshot_name).json())
self.check_equal(self.get("/api/snapshots/" + snapshot_name).status_code, 200)
@@ -26,8 +25,8 @@ class SnapshotsAPITestCreateShowEmpty(APITest):
self.check_equal(self.get("/api/snapshots/" + self.random_name()).status_code, 404)
# create snapshot with duplicate name
resp = self.post("/api/snapshots", json=snapshot_desc)
self.check_equal(resp.status_code, 400)
resp = self.post_task("/api/snapshots", json=snapshot_desc)
self.check_equal(resp.json()['State'], 3)
class SnapshotsAPITestCreateFromRefs(APITest):
@@ -37,9 +36,9 @@ class SnapshotsAPITestCreateFromRefs(APITest):
"""
def check(self):
snapshot_name = self.random_name()
snapshot_desc = {u'Description': u'fun snapshot',
u'Name': snapshot_name,
u'SourceSnapshots': [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)
@@ -47,9 +46,11 @@ class SnapshotsAPITestCreateFromRefs(APITest):
# create empty snapshot
empty_snapshot_name = self.random_name()
resp = self.post("/api/snapshots", json={"Name": empty_snapshot_name})
self.check_equal(resp.status_code, 201)
self.check_equal(resp.json()['Description'], 'Created as empty')
resp = self.post_task("/api/snapshots", json={"Name": empty_snapshot_name})
self.check_equal(resp.json()['State'], 2)
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()
@@ -57,16 +58,18 @@ class SnapshotsAPITestCreateFromRefs(APITest):
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)
self.check_equal(self.post("/api/repos/" + repo_name + "/file/" + d).status_code, 200)
self.check_equal(self.post_task("/api/repos/" + repo_name + "/file/" + d).json()['State'], 2)
# 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]
resp = self.post("/api/snapshots", json=snapshot)
self.check_equal(resp.status_code, 201)
resp = self.post_task("/api/snapshots", json=snapshot)
self.check_equal(resp.json()['State'], 2)
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())
@@ -75,10 +78,10 @@ class SnapshotsAPITestCreateFromRefs(APITest):
self.check_equal(resp.json(), ["Pi386 libboost-program-options-dev 1.49.0.1 918d2f433384e378"])
# create snapshot with unreferenced package
resp = self.post("/api/snapshots", json={
resp = 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_equal(resp.status_code, 404)
self.check_equal(resp.json()['State'], 3)
# list snapshots
resp = self.get("/api/snapshots", params={"sort": "time"})
@@ -96,8 +99,8 @@ class SnapshotsAPITestCreateFromRepo(APITest):
snapshot_name = self.random_name()
self.check_equal(self.post("/api/repos", json={"Name": repo_name}).status_code, 201)
resp = self.post("/api/repos/" + repo_name + '/snapshots', json={'Name': snapshot_name})
self.check_equal(resp.status_code, 201)
resp = self.post_task("/api/repos/" + repo_name + '/snapshots', json={'Name': snapshot_name})
self.check_equal(resp.json()['State'], 2)
self.check_equal([],
self.get("/api/snapshots/" + snapshot_name + "/packages", params={"format": "details"}).json())
@@ -106,28 +109,28 @@ class SnapshotsAPITestCreateFromRepo(APITest):
self.check_equal(self.upload("/api/files/" + d,
"libboost-program-options-dev_1.49.0.1_i386.deb").status_code, 200)
self.check_equal(self.post("/api/repos/" + repo_name + "/file/" + d).status_code, 200)
self.check_equal(self.post_task("/api/repos/" + repo_name + "/file/" + d).json()['State'], 2)
resp = self.post("/api/repos/" + repo_name + '/snapshots', json={'Name': snapshot_name})
self.check_equal(resp.status_code, 201)
resp = self.post_task("/api/repos/" + repo_name + '/snapshots', json={'Name': snapshot_name})
self.check_equal(resp.json()['State'], 2)
self.check_equal(self.get("/api/snapshots/" + snapshot_name).status_code, 200)
self.check_subset({u'Architecture': 'i386',
u'Package': 'libboost-program-options-dev',
u'Version': '1.49.0.1',
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({u'Architecture': 'i386',
u'Package': 'libboost-program-options-dev',
u'Version': '1.49.0.1',
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
resp = self.post("/api/repos/" + repo_name + '/snapshots', json={'Name': snapshot_name})
self.check_equal(resp.status_code, 400)
resp = self.post_task("/api/repos/" + repo_name + '/snapshots', json={'Name': snapshot_name})
self.check_equal(resp.json()['State'], 3)
class SnapshotsAPITestCreateUpdate(APITest):
@@ -136,16 +139,16 @@ class SnapshotsAPITestCreateUpdate(APITest):
"""
def check(self):
snapshot_name = self.random_name()
snapshot_desc = {u'Description': u'fun snapshot',
u'Name': snapshot_name}
snapshot_desc = {'Description': 'fun snapshot',
'Name': snapshot_name}
resp = self.post("/api/snapshots", json=snapshot_desc)
self.check_equal(resp.status_code, 201)
resp = self.post_task("/api/snapshots", json=snapshot_desc)
self.check_equal(resp.json()['State'], 2)
new_snapshot_name = self.random_name()
resp = self.put("/api/snapshots/" + snapshot_name, json={'Name': new_snapshot_name,
'Description': 'New description'})
self.check_equal(resp.status_code, 200)
resp = self.put_task("/api/snapshots/" + snapshot_name, json={'Name': new_snapshot_name,
'Description': 'New description'})
self.check_equal(resp.json()['State'], 2)
resp = self.get("/api/snapshots/" + new_snapshot_name)
self.check_equal(resp.status_code, 200)
@@ -153,9 +156,9 @@ class SnapshotsAPITestCreateUpdate(APITest):
"Description": "New description"}, resp.json())
# duplicate name
resp = self.put("/api/snapshots/" + new_snapshot_name, json={'Name': new_snapshot_name,
'Description': 'New description'})
self.check_equal(resp.status_code, 409)
resp = self.put_task("/api/snapshots/" + new_snapshot_name, json={'Name': new_snapshot_name,
'Description': 'New description'})
self.check_equal(resp.json()['State'], 3)
# missing snapshot
resp = self.put("/api/snapshots/" + snapshot_name, json={})
@@ -168,40 +171,46 @@ class SnapshotsAPITestCreateDelete(APITest):
"""
def check(self):
snapshot_name = self.random_name()
snapshot_desc = {u'Description': u'fun snapshot',
u'Name': snapshot_name}
snapshot_desc = {'Description': 'fun snapshot',
'Name': snapshot_name}
# deleting unreferenced snapshot
resp = self.post("/api/snapshots", json=snapshot_desc)
self.check_equal(resp.status_code, 201)
resp = self.post_task("/api/snapshots", json=snapshot_desc)
self.check_equal(resp.json()['State'], 2)
self.check_equal(self.delete("/api/snapshots/" + snapshot_name).status_code, 200)
self.check_equal(self.delete_task("/api/snapshots/" + snapshot_name).json()['State'], 2)
self.check_equal(self.get("/api/snapshots/" + snapshot_name).status_code, 404)
# deleting referenced snapshot
snap1, snap2 = self.random_name(), self.random_name()
self.check_equal(self.post("/api/snapshots", json={"Name": snap1}).status_code, 201)
self.check_equal(self.post("/api/snapshots", json={"Name": snap2, "SourceSnapshots": [snap1]}).status_code, 201)
self.check_equal(self.post_task("/api/snapshots", json={"Name": snap1}).json()['State'], 2)
self.check_equal(
self.post_task(
"/api/snapshots", json={"Name": snap2, "SourceSnapshots": [snap1]}
).json()['State'], 2
)
self.check_equal(self.delete("/api/snapshots/" + snap1).status_code, 409)
self.check_equal(self.delete_task("/api/snapshots/" + snap1).json()['State'], 3)
self.check_equal(self.get("/api/snapshots/" + snap1).status_code, 200)
self.check_equal(self.delete("/api/snapshots/" + snap1, params={"force": "1"}).status_code, 200)
self.check_equal(self.delete_task("/api/snapshots/" + snap1, params={"force": "1"}).json()['State'], 2)
self.check_equal(self.get("/api/snapshots/" + snap1).status_code, 404)
# deleting published snapshot
resp = self.post("/api/publish",
json={
"SourceKind": "snapshot",
"Distribution": "trusty",
"Architectures": ["i386"],
"Sources": [{"Name": snap2}],
"Signing": DefaultSigningOptions,
})
self.check_equal(resp.status_code, 201)
resp = self.post_task(
"/api/publish",
json={
"SourceKind": "snapshot",
"Distribution": "trusty",
"Architectures": ["i386"],
"Sources": [{"Name": snap2}],
"Signing": DefaultSigningOptions,
}
)
self.check_equal(resp.json()['State'], 2)
self.check_equal(self.delete("/api/snapshots/" + snap2).status_code, 409)
self.check_equal(self.delete("/api/snapshots/" + snap2, params={"force": "1"}).status_code, 409)
self.check_equal(self.delete_task("/api/snapshots/" + snap2).json()['State'], 3)
self.check_equal(self.delete_task("/api/snapshots/" + snap2, params={"force": "1"}).json()['State'], 3)
class SnapshotsAPITestSearch(APITest):
@@ -218,10 +227,10 @@ class SnapshotsAPITestSearch(APITest):
self.check_equal(self.upload("/api/files/" + d,
"libboost-program-options-dev_1.49.0.1_i386.deb").status_code, 200)
self.check_equal(self.post("/api/repos/" + repo_name + "/file/" + d).status_code, 200)
self.check_equal(self.post_task("/api/repos/" + repo_name + "/file/" + d).json()['State'], 2)
resp = self.post("/api/repos/" + repo_name + '/snapshots', json={'Name': snapshot_name})
self.check_equal(resp.status_code, 201)
resp = self.post_task("/api/repos/" + repo_name + '/snapshots', json={'Name': snapshot_name})
self.check_equal(resp.json()['State'], 2)
resp = self.get("/api/snapshots/" + snapshot_name + "/packages",
params={"q": "libboost-program-options-dev", "format": "details"})
@@ -242,8 +251,8 @@ class SnapshotsAPITestDiff(APITest):
GET /api/snapshot/:name/diff/:name2
"""
def check(self):
repos = [self.random_name() for x in xrange(2)]
snapshots = [self.random_name() for x in xrange(2)]
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)
@@ -252,13 +261,13 @@ class SnapshotsAPITestDiff(APITest):
self.check_equal(self.upload("/api/files/" + d,
"libboost-program-options-dev_1.49.0.1_i386.deb").status_code, 200)
self.check_equal(self.post("/api/repos/" + repo_name + "/file/" + d).status_code, 200)
self.check_equal(self.post_task("/api/repos/" + repo_name + "/file/" + d).json()['State'], 2)
resp = self.post("/api/repos/" + repo_name + '/snapshots', json={'Name': snapshots[0]})
self.check_equal(resp.status_code, 201)
resp = self.post_task("/api/repos/" + repo_name + '/snapshots', json={'Name': snapshots[0]})
self.check_equal(resp.json()['State'], 2)
resp = self.post("/api/snapshots", json={'Name': snapshots[1]})
self.check_equal(resp.status_code, 201)
resp = self.post_task("/api/snapshots", json={'Name': snapshots[1]})
self.check_equal(resp.json()['State'], 2)
resp = self.get("/api/snapshots/" + snapshots[0] + "/diff/" + snapshots[1])
self.check_equal(resp.status_code, 200)
system/t12_api/systemd_handover.py
+4 -2
View File
@@ -1,5 +1,7 @@
import requests_unixsocket
import urllib
import urllib.error
import urllib.parse
import urllib.request
import os
import os.path
@@ -42,5 +44,5 @@ class SystemdAPIHandoverTest(BaseTest):
print("Skipping test as we failed to setup a listener.")
return
session = requests_unixsocket.Session()
r = session.get('http+unix://%s/api/version' % urllib.quote(self.socket_path, safe=''))
r = session.get('http+unix://%s/api/version' % urllib.parse.quote(self.socket_path, safe=''))
self.check_equal(r.json(), {'Version': os.environ['APTLY_VERSION']})
system/t12_api/tasks.py
+106
View File
@@ -0,0 +1,106 @@
from api_lib import APITest
from .publish import DefaultSigningOptions
class TaskAPITestParallelTasks(APITest):
"""
GET /api/tasks, GET /api/tasks/:id/wait, GET /api/tasks-wait
"""
def _create_mirror(self, dist):
mirror_name = self.random_name()
mirror_desc = {'Name': mirror_name,
'ArchiveURL': 'https://packagecloud.io/varnishcache/varnish30/debian/',
'Distribution': dist,
'Components': ['main']}
mirror_desc['IgnoreSignatures'] = True
resp = self.post("/api/mirrors", json=mirror_desc)
self.check_equal(resp.status_code, 201)
resp = self.put("/api/mirrors/" + mirror_name, json=mirror_desc, params={'_async': True})
self.check_equal(resp.status_code, 202)
# check that two mirror updates cannot run at the same time
resp2 = self.put("/api/mirrors/" + mirror_name, json=mirror_desc, params={'_async': True})
self.check_equal(resp2.status_code, 409)
return resp.json()['ID'], mirror_name
def _create_repo(self):
repo_name = self.random_name()
distribution = self.random_name()
self.check_equal(self.post("/api/repos",
json={
"Name": repo_name,
"Comment": "fun repo",
"DefaultDistribution": distribution
}).status_code, 201)
d = self.random_name()
self.check_equal(
self.upload("/api/files/" + d, "pyspi_0.6.1-1.3.dsc",
"pyspi_0.6.1-1.3.diff.gz",
"pyspi_0.6.1.orig.tar.gz").status_code, 200)
resp = self.post("/api/repos/" + repo_name + "/file/" + d, params={'_async': True})
self.check_equal(resp.status_code, 202)
return resp.json()['ID'], repo_name
def _wait_for_task(self, task_id):
uri = "/api/tasks/%d/wait" % int(task_id)
resp = self.get(uri)
self.check_equal(resp.status_code, 200)
self.check_equal(resp.json()['State'], 2)
def _wait_for_all_tasks(self):
resp = self.get("/api/tasks-wait")
self.check_equal(resp.status_code, 200)
def _snapshot(self, res_type, name):
uri = "/api/%s/%s/snapshots" % (res_type, name)
resp = self.post(uri, json={"Name": name}, params={'_async': True})
self.check_equal(resp.status_code, 202)
return resp.json()['ID']
def _publish(self, source_kind, name):
resp = self.post("/api/publish",
json={
"SourceKind": source_kind,
"Sources": [{"Name": name}],
"Signing": DefaultSigningOptions,
}, params={'_async': True})
self.check_equal(resp.status_code, 202)
return resp.json()['ID']
def check(self):
publish_task_ids = []
mirror_task_list = []
for mirror_dist in ['squeeze', 'jessie']:
mirror_task_id, mirror_name = self._create_mirror(mirror_dist)
mirror_task_list.append((mirror_task_id, mirror_name))
repo_task_id, repo_name = self._create_repo()
self._wait_for_task(repo_task_id)
resp = self.delete("/api/tasks/%d" % repo_task_id)
self.check_equal(resp.status_code, 200)
resp = self.get("/api/tasks/%d" % repo_task_id)
self.check_equal(resp.status_code, 404)
repo_snap_task_id = self._snapshot('repos', repo_name)
self._wait_for_task(repo_snap_task_id)
publish_task_ids.append(self._publish('snapshot', repo_name))
for mirror_task_id, mirror_name in mirror_task_list:
self._wait_for_task(mirror_task_id)
mirror_snap_task_id = self._snapshot('mirrors', mirror_name)
self._wait_for_task(mirror_snap_task_id)
publish_task_ids.append(self._publish('snapshot', mirror_name))
self._wait_for_all_tasks()
for publish_task_id in publish_task_ids:
resp = self.get("/api/tasks/%d" % publish_task_id)
self.check_equal(resp.status_code, 200)
self.check_equal(resp.json()['State'], 2)
system/t12_api/unix_socket.py
+4 -2
View File
@@ -1,7 +1,9 @@
import requests_unixsocket
import time
import os
import urllib
import urllib.error
import urllib.parse
import urllib.request
from lib import BaseTest
@@ -32,7 +34,7 @@ class UnixSocketAPITest(BaseTest):
"""
def check(self):
session = requests_unixsocket.Session()
r = session.get('http+unix://%s/api/version' % urllib.quote(UnixSocketAPITest.socket_path, safe=''))
r = session.get('http+unix://%s/api/version' % urllib.parse.quote(UnixSocketAPITest.socket_path, safe=''))
# Just needs to come back, we actually don't care much about the code.
# Only needs to verify that the socket is actually responding.
self.check_equal(r.json(), {'Version': os.environ['APTLY_VERSION']})