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

improve api tests and error output

Browse Source 
show only relevant aptly logs if a test fails.
for async tasks, show task output, as it contains the error message.
This commit is contained in:
André Roth
2024-07-16 19:48:04 +02:00
parent 8029305d32
commit 440c3debdc
11 changed files with 278 additions and 175 deletions
Download Patch File Download Diff File Expand all files Collapse all files
system/api_lib.py
+46 -1
View File
@@ -5,6 +5,7 @@ import random
import shutil import shutil
import string import string
import time import time
import tempfile
from lib import BaseTest from lib import BaseTest
@@ -18,11 +19,32 @@ TASK_SUCCEEDED = 2
TASK_FAILED = 3 TASK_FAILED = 3
class AptlyStream:
def __init__(self):
self.tmp_file = tempfile.NamedTemporaryFile(delete=False)
self.read_pos = 0
def fileno(self):
return self.tmp_file.fileno()
def get_contents(self):
self.tmp_file.seek(self.read_pos, 0)
return self.tmp_file.read().decode("utf-8")
def close(self):
self.tmp_file.close()
def clear(self):
self.read_pos = self.tmp_file.tell()
class APITest(BaseTest): class APITest(BaseTest):
""" """
BaseTest + testing aptly API BaseTest + testing aptly API
""" """
aptly_server = None aptly_server = None
aptly_out = None
debugOutput = True
base_url = "127.0.0.1:8765" base_url = "127.0.0.1:8765"
configOverride = { configOverride = {
"FileSystemPublishEndpoints": { "FileSystemPublishEndpoints": {
@@ -40,13 +62,20 @@ class APITest(BaseTest):
if APITest.aptly_server is None: if APITest.aptly_server is None:
super(APITest, self).prepare() super(APITest, self).prepare()
APITest.aptly_out = AptlyStream()
configPath = os.path.join(os.environ["HOME"], self.aptlyConfigFile) configPath = os.path.join(os.environ["HOME"], self.aptlyConfigFile)
APITest.aptly_server = self._start_process(f"aptly api serve -no-lock -config={configPath} -listen={self.base_url}",) APITest.aptly_server = self._start_process(f"aptly api serve -no-lock -config={configPath} -listen={self.base_url}", stdout=APITest.aptly_out, stderr=APITest.aptly_out)
time.sleep(1) time.sleep(1)
else:
APITest.aptly_out.clear()
if os.path.exists(os.path.join(os.environ["HOME"], self.aptlyDir, "upload")): if os.path.exists(os.path.join(os.environ["HOME"], self.aptlyDir, "upload")):
shutil.rmtree(os.path.join(os.environ["HOME"], self.aptlyDir, "upload")) shutil.rmtree(os.path.join(os.environ["HOME"], self.aptlyDir, "upload"))
def debug_output(self):
return APITest.aptly_out.get_contents()
def run(self): def run(self):
pass pass
@@ -79,6 +108,20 @@ class APITest(BaseTest):
return self.get("/api/tasks/" + str(_id)) return self.get("/api/tasks/" + str(_id))
def check_task(self, task):
self.check_equal(task.status_code, 200)
if task.json()['State'] != TASK_SUCCEEDED:
resp2 = self.get("/api/tasks/" + str(task.json()['ID']) + "/output")
raise Exception(f"task failed: {resp2.text}")
def check_task_fail(self, task, expected_output=None):
self.check_equal(task.status_code, 200)
if task.json()['State'] == TASK_SUCCEEDED:
raise Exception("task expected to fail")
if expected_output:
resp = self.get("/api/tasks/" + str(task.json()['ID']) + "/output")
self.check_equal(resp.text, expected_output)
def put(self, uri, *args, **kwargs): def put(self, uri, *args, **kwargs):
if "json" in kwargs: if "json" in kwargs:
kwargs["data"] = json.dumps(kwargs.pop("json")) kwargs["data"] = json.dumps(kwargs.pop("json"))
@@ -142,6 +185,8 @@ class APITest(BaseTest):
cls.aptly_server.terminate() cls.aptly_server.terminate()
cls.aptly_server.wait() cls.aptly_server.wait()
cls.aptly_server = None cls.aptly_server = None
if APITest.aptly_out is not None:
APITest.aptly_out.close()
def random_name(self): def random_name(self):
return ''.join(random.choice(string.ascii_letters + string.digits) for _ in range(15)) return ''.join(random.choice(string.ascii_letters + string.digits) for _ in range(15))
system/lib.py
+5
View File
@@ -128,6 +128,7 @@ class BaseTest(object):
requiresGPG2 = False requiresGPG2 = False
requiresDot = False requiresDot = False
sortOutput = False sortOutput = False
debugOutput = False
aptlyDir = ".aptly" aptlyDir = ".aptly"
aptlyConfigFile = ".aptly.conf" aptlyConfigFile = ".aptly.conf"
@@ -177,6 +178,10 @@ class BaseTest(object):
try: try:
self.run() self.run()
self.check() self.check()
except Exception as exc:
if self.debugOutput:
print(f"API log:\n{self.debug_output()}")
raise exc
finally: finally:
self.teardown() self.teardown()
system/run.py
+2 -4
View File
@@ -153,10 +153,8 @@ def run(include_long_tests=False, capture_results=False, tests=None, filters=Non
for (test, t, typ, val, tb, testModule) in fails: for (test, t, typ, val, tb, testModule) in fails:
doc = t.__doc__ or '' doc = t.__doc__ or ''
print("%s:%s %s" % (test, t.__class__.__name__, print(" - %s:%s %s" % (test, t.__class__.__name__,
testModule.__name__ + ": " + doc.strip())) testModule.__name__ + ": " + doc.strip()))
traceback.print_exception(typ, val, tb)
print("=" * 60)
sys.exit(1) sys.exit(1)
system/t12_api/gpg.py
+8 -3
View File
@@ -7,11 +7,16 @@ from api_lib import APITest
def check_gpgkey_exists(gpg_key, keyring): def check_gpgkey_exists(gpg_key, keyring):
subprocess.check_call([ p = subprocess.Popen([
"gpg", "--no-default-keyring", "gpg", "--no-default-keyring",
"--keyring", keyring, "--keyring", keyring,
"--fingerprint", gpg_key, "--fingerprint", gpg_key],
]) stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
p.communicate()
if p.returncode != 0:
raise Exception("gpg key does not exists")
class GPGAPITestAddKey(APITest): class GPGAPITestAddKey(APITest):
system/t12_api/mirrors.py
+4 -11
View File
@@ -1,4 +1,4 @@
from api_lib import TASK_SUCCEEDED, APITest from api_lib import APITest
class MirrorsAPITestCreateShow(APITest): class MirrorsAPITestCreateShow(APITest):
@@ -58,11 +58,8 @@ class MirrorsAPITestCreateUpdate(APITest):
mirror_desc["Name"] = self.random_name() mirror_desc["Name"] = self.random_name()
resp = self.put_task("/api/mirrors/" + mirror_name, json=mirror_desc) resp = self.put_task("/api/mirrors/" + mirror_name, json=mirror_desc)
self.check_equal(resp.status_code, 200) self.check_task(resp)
_id = resp.json()['ID'] _id = resp.json()['ID']
if resp.json()["State"] != TASK_SUCCEEDED:
resp = self.get("/api/tasks/" + str(_id) + "/output")
raise Exception("task failed: " + str(resp.json()))
resp = self.get("/api/tasks/" + str(_id) + "/detail") resp = self.get("/api/tasks/" + str(_id) + "/detail")
self.check_equal(resp.status_code, 200) self.check_equal(resp.status_code, 200)
@@ -95,7 +92,7 @@ class MirrorsAPITestCreateDelete(APITest):
self.check_equal(resp.status_code, 201) self.check_equal(resp.status_code, 201)
resp = self.delete_task("/api/mirrors/" + mirror_name) resp = self.delete_task("/api/mirrors/" + mirror_name)
self.check_equal(resp.json()['State'], TASK_SUCCEEDED) self.check_task(resp)
class MirrorsAPITestCreateList(APITest): class MirrorsAPITestCreateList(APITest):
@@ -154,8 +151,4 @@ class MirrorsAPITestSkipArchitectureCheck(APITest):
mirror_desc = {'Name': mirror_name, mirror_desc = {'Name': mirror_name,
'IgnoreSignatures': True} 'IgnoreSignatures': True}
resp = self.put_task("/api/mirrors/" + mirror_name, json=mirror_desc) resp = self.put_task("/api/mirrors/" + mirror_name, json=mirror_desc)
self.check_equal(resp.status_code, 200) self.check_task(resp)
_id = resp.json()['ID']
if resp.json()["State"] != TASK_SUCCEEDED:
resp = self.get("/api/tasks/" + str(_id) + "/output")
raise Exception("task failed: " + str(resp.json()))
system/t12_api/packages.py
+2 -2
View File
@@ -2,7 +2,7 @@ import urllib.error
import urllib.parse import urllib.parse
import urllib.request import urllib.request
from api_lib import TASK_SUCCEEDED, APITest from api_lib import APITest
class PackagesAPITestShow(APITest): class PackagesAPITestShow(APITest):
@@ -19,7 +19,7 @@ class PackagesAPITestShow(APITest):
"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) "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_task("/api/repos/" + repo_name + "/file/" + d) resp = self.post_task("/api/repos/" + repo_name + "/file/" + d)
self.check_equal(resp.json()['State'], TASK_SUCCEEDED) self.check_task(resp)
# get information about package # get information about package
pyspi_json = { pyspi_json = {
system/t12_api/publish.py
+88 -66
View File
@@ -27,11 +27,12 @@ class PublishAPITestRepo(APITest):
"pyspi_0.6.1-1.3.diff.gz", "pyspi_0.6.1.orig.tar.gz", "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) "pyspi-0.6.1-1.3.stripped.dsc").status_code, 200)
self.check_equal(self.post_task("/api/repos/" + repo_name + "/file/" + d).json()['State'], TASK_SUCCEEDED) task = self.post_task("/api/repos/" + repo_name + "/file/" + d)
self.check_task(task)
# publishing under prefix, default distribution # publishing under prefix, default distribution
prefix = self.random_name() prefix = self.random_name()
resp = self.post_task( task = self.post_task(
"/api/publish/" + prefix, "/api/publish/" + prefix,
json={ json={
"SourceKind": "local", "SourceKind": "local",
@@ -39,6 +40,7 @@ class PublishAPITestRepo(APITest):
"Signing": DefaultSigningOptions, "Signing": DefaultSigningOptions,
} }
) )
self.check_task(task)
repo_expected = { repo_expected = {
'AcquireByHash': False, 'AcquireByHash': False,
'Architectures': ['i386', 'source'], 'Architectures': ['i386', 'source'],
@@ -56,8 +58,6 @@ class PublishAPITestRepo(APITest):
'Storage': '', 'Storage': '',
'Suite': ''} 'Suite': ''}
self.check_equal(resp.json()['State'], TASK_SUCCEEDED)
all_repos = self.get("/api/publish") all_repos = self.get("/api/publish")
self.check_equal(all_repos.status_code, 200) self.check_equal(all_repos.status_code, 200)
self.check_in(repo_expected, all_repos.json()) self.check_in(repo_expected, all_repos.json())
@@ -74,7 +74,7 @@ class PublishAPITestRepo(APITest):
# publishing under root, custom distribution, architectures # publishing under root, custom distribution, architectures
distribution = self.random_name() distribution = self.random_name()
resp = self.post_task( task = self.post_task(
"/api/publish/:.", "/api/publish/:.",
json={ json={
"SourceKind": "local", "SourceKind": "local",
@@ -84,7 +84,7 @@ class PublishAPITestRepo(APITest):
"Architectures": ["i386", "amd64"], "Architectures": ["i386", "amd64"],
} }
) )
self.check_equal(resp.json()['State'], TASK_SUCCEEDED) self.check_task(task)
repo2_expected = { repo2_expected = {
'AcquireByHash': False, 'AcquireByHash': False,
'Architectures': ['amd64', 'i386'], 'Architectures': ['amd64', 'i386'],
@@ -138,12 +138,14 @@ class PublishSnapshotAPITest(APITest):
self.check_equal(self.upload("/api/files/" + d, 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_task("/api/repos/" + repo_name + "/file/" + d).json()['State'], TASK_SUCCEEDED) task = self.post_task("/api/repos/" + repo_name + "/file/" + d)
self.check_task(task)
self.check_equal(self.post_task("/api/repos/" + repo_name + '/snapshots', json={'Name': snapshot_name}).json()['State'], TASK_SUCCEEDED) task = self.post_task("/api/repos/" + repo_name + '/snapshots', json={'Name': snapshot_name})
self.check_task(task)
prefix = self.random_name() prefix = self.random_name()
resp = self.post_task( task = self.post_task(
"/api/publish/" + prefix, "/api/publish/" + prefix,
json={ json={
"AcquireByHash": True, "AcquireByHash": True,
@@ -157,9 +159,9 @@ class PublishSnapshotAPITest(APITest):
"Label": "fun", "Label": "fun",
} }
) )
self.check_equal(resp.json()['State'], TASK_SUCCEEDED) self.check_task(task)
_id = resp.json()['ID'] _id = task.json()['ID']
resp = self.get("/api/tasks/" + str(_id) + "/detail") resp = self.get("/api/tasks/" + str(_id) + "/detail")
self.check_equal(resp.json()['RemainingNumberOfPackages'], 0) self.check_equal(resp.json()['RemainingNumberOfPackages'], 0)
self.check_equal(resp.json()['TotalNumberOfPackages'], 1) self.check_equal(resp.json()['TotalNumberOfPackages'], 1)
@@ -213,10 +215,11 @@ class PublishUpdateAPITestRepo(APITest):
"pyspi_0.6.1-1.3.dsc", "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.diff.gz", "pyspi_0.6.1.orig.tar.gz",
"pyspi-0.6.1-1.3.stripped.dsc").status_code, 200) "pyspi-0.6.1-1.3.stripped.dsc").status_code, 200)
self.check_equal(self.post_task("/api/repos/" + repo_name + "/file/" + d).json()['State'], TASK_SUCCEEDED) task = self.post_task("/api/repos/" + repo_name + "/file/" + d)
self.check_task(task)
prefix = self.random_name() prefix = self.random_name()
resp = self.post_task( task = self.post_task(
"/api/publish/" + prefix, "/api/publish/" + prefix,
json={ json={
"Architectures": ["i386", "source"], "Architectures": ["i386", "source"],
@@ -225,8 +228,7 @@ class PublishUpdateAPITestRepo(APITest):
"Signing": DefaultSigningOptions, "Signing": DefaultSigningOptions,
} }
) )
self.check_task(task)
self.check_equal(resp.json()['State'], TASK_SUCCEEDED)
self.check_not_exists( self.check_not_exists(
"public/" + prefix + "/pool/main/b/boost-defaults/libboost-program-options-dev_1.49.0.1_i386.deb") "public/" + prefix + "/pool/main/b/boost-defaults/libboost-program-options-dev_1.49.0.1_i386.deb")
@@ -236,20 +238,22 @@ class PublishUpdateAPITestRepo(APITest):
d = self.random_name() d = self.random_name()
self.check_equal(self.upload("/api/files/" + d, 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_task("/api/repos/" + repo_name + "/file/" + d).json()['State'], TASK_SUCCEEDED) task = self.post_task("/api/repos/" + repo_name + "/file/" + d)
self.check_task(task)
self.check_equal(self.delete_task("/api/repos/" + repo_name + "/packages/", task = self.delete_task("/api/repos/" + repo_name + "/packages/",
json={"PackageRefs": ['Psource pyspi 0.6.1-1.4 f8f1daa806004e89']}).json()['State'], TASK_SUCCEEDED) json={"PackageRefs": ['Psource pyspi 0.6.1-1.4 f8f1daa806004e89']})
self.check_task(task)
# Update and switch AcquireByHash on. # Update and switch AcquireByHash on.
resp = self.put_task( task = self.put_task(
"/api/publish/" + prefix + "/wheezy", "/api/publish/" + prefix + "/wheezy",
json={ json={
"AcquireByHash": True, "AcquireByHash": True,
"Signing": DefaultSigningOptions, "Signing": DefaultSigningOptions,
} }
) )
self.check_equal(resp.json()['State'], TASK_SUCCEEDED) self.check_task(task)
repo_expected = { repo_expected = {
'AcquireByHash': True, 'AcquireByHash': True,
'Architectures': ['i386', 'source'], 'Architectures': ['i386', 'source'],
@@ -279,7 +283,8 @@ class PublishUpdateAPITestRepo(APITest):
self.check_not_exists( self.check_not_exists(
"public/" + prefix + "/pool/main/p/pyspi/pyspi-0.6.1-1.3.stripped.dsc") "public/" + prefix + "/pool/main/p/pyspi/pyspi-0.6.1-1.3.stripped.dsc")
self.check_equal(self.delete_task("/api/publish/" + prefix + "/wheezy").json()['State'], TASK_SUCCEEDED) task = self.delete_task("/api/publish/" + prefix + "/wheezy")
self.check_task(task)
self.check_not_exists("public/" + prefix + "dists/") self.check_not_exists("public/" + prefix + "dists/")
@@ -300,10 +305,11 @@ class PublishConcurrentUpdateAPITestRepo(APITest):
"pyspi_0.6.1-1.3.dsc", "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.diff.gz", "pyspi_0.6.1.orig.tar.gz",
"pyspi-0.6.1-1.3.stripped.dsc").status_code, 200) "pyspi-0.6.1-1.3.stripped.dsc").status_code, 200)
self.check_equal(self.post_task("/api/repos/" + repo_name + "/file/" + d).json()['State'], TASK_SUCCEEDED) task = self.post_task("/api/repos/" + repo_name + "/file/" + d)
self.check_task(task)
prefix = self.random_name() prefix = self.random_name()
resp = self.post_task( task = self.post_task(
"/api/publish/" + prefix, "/api/publish/" + prefix,
json={ json={
"Architectures": ["i386", "source"], "Architectures": ["i386", "source"],
@@ -312,8 +318,7 @@ class PublishConcurrentUpdateAPITestRepo(APITest):
"Signing": DefaultSigningOptions, "Signing": DefaultSigningOptions,
} }
) )
self.check_task(task)
self.check_equal(resp.json()['State'], TASK_SUCCEEDED)
self.check_not_exists( self.check_not_exists(
"public/" + prefix + "/pool/main/b/boost-defaults/libboost-program-options-dev_1.49.0.1_i386.deb") "public/" + prefix + "/pool/main/b/boost-defaults/libboost-program-options-dev_1.49.0.1_i386.deb")
@@ -323,10 +328,12 @@ class PublishConcurrentUpdateAPITestRepo(APITest):
d = self.random_name() d = self.random_name()
self.check_equal(self.upload("/api/files/" + d, 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_task("/api/repos/" + repo_name + "/file/" + d).json()['State'], TASK_SUCCEEDED) task = self.post_task("/api/repos/" + repo_name + "/file/" + d)
self.check_task(task)
self.check_equal(self.delete_task("/api/repos/" + repo_name + "/packages/", task = self.delete_task("/api/repos/" + repo_name + "/packages/",
json={"PackageRefs": ['Psource pyspi 0.6.1-1.4 f8f1daa806004e89']}).json()['State'], TASK_SUCCEEDED) json={"PackageRefs": ['Psource pyspi 0.6.1-1.4 f8f1daa806004e89']})
self.check_task(task)
def _do_update(result, index): def _do_update(result, index):
resp = self.put_task( resp = self.put_task(
@@ -379,7 +386,8 @@ class PublishConcurrentUpdateAPITestRepo(APITest):
self.check_not_exists( self.check_not_exists(
"public/" + prefix + "/pool/main/p/pyspi/pyspi-0.6.1-1.3.stripped.dsc") "public/" + prefix + "/pool/main/p/pyspi/pyspi-0.6.1-1.3.stripped.dsc")
self.check_equal(self.delete_task("/api/publish/" + prefix + "/wheezy").json()['State'], TASK_SUCCEEDED) task = self.delete_task("/api/publish/" + prefix + "/wheezy")
self.check_task(task)
self.check_not_exists("public/" + prefix + "dists/") self.check_not_exists("public/" + prefix + "dists/")
@@ -400,18 +408,18 @@ class PublishUpdateSkipCleanupAPITestRepo(APITest):
"pyspi_0.6.1-1.3.dsc", "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.diff.gz", "pyspi_0.6.1.orig.tar.gz",
"pyspi-0.6.1-1.3.stripped.dsc").status_code, 200) "pyspi-0.6.1-1.3.stripped.dsc").status_code, 200)
self.check_equal(self.post_task("/api/repos/" + repo_name + "/file/" + d).json()['State'], TASK_SUCCEEDED) task = self.post_task("/api/repos/" + repo_name + "/file/" + d)
self.check_task(task)
prefix = self.random_name() prefix = self.random_name()
resp = self.post_task("/api/publish/" + prefix, task = self.post_task("/api/publish/" + prefix,
json={ json={
"Architectures": ["i386", "source"], "Architectures": ["i386", "source"],
"SourceKind": "local", "SourceKind": "local",
"Sources": [{"Name": repo_name}], "Sources": [{"Name": repo_name}],
"Signing": DefaultSigningOptions, "Signing": DefaultSigningOptions,
}) })
self.check_task(task)
self.check_equal(resp.json()['State'], TASK_SUCCEEDED)
self.check_not_exists( self.check_not_exists(
"public/" + prefix + "/pool/main/b/boost-defaults/libboost-program-options-dev_1.49.0.1_i386.deb") "public/" + prefix + "/pool/main/b/boost-defaults/libboost-program-options-dev_1.49.0.1_i386.deb")
@@ -420,7 +428,7 @@ class PublishUpdateSkipCleanupAPITestRepo(APITest):
# Publish two repos, so that deleting one while skipping cleanup will # Publish two repos, so that deleting one while skipping cleanup will
# not delete the whole prefix. # not delete the whole prefix.
resp = self.post_task("/api/publish/" + prefix, task = self.post_task("/api/publish/" + prefix,
json={ json={
"Architectures": ["i386", "source"], "Architectures": ["i386", "source"],
"Distribution": "otherdist", "Distribution": "otherdist",
@@ -428,23 +436,24 @@ class PublishUpdateSkipCleanupAPITestRepo(APITest):
"Sources": [{"Name": repo_name}], "Sources": [{"Name": repo_name}],
"Signing": DefaultSigningOptions, "Signing": DefaultSigningOptions,
}) })
self.check_task(task)
self.check_equal(resp.json()['State'], TASK_SUCCEEDED)
d = self.random_name() d = self.random_name()
self.check_equal(self.upload("/api/files/" + d, 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_task("/api/repos/" + repo_name + "/file/" + d).json()['State'], TASK_SUCCEEDED) task = self.post_task("/api/repos/" + repo_name + "/file/" + d)
self.check_task(task)
self.check_equal(self.delete_task("/api/repos/" + repo_name + "/packages/", task = self.delete_task("/api/repos/" + repo_name + "/packages/",
json={"PackageRefs": ['Psource pyspi 0.6.1-1.4 f8f1daa806004e89']}).json()['State'], TASK_SUCCEEDED) json={"PackageRefs": ['Psource pyspi 0.6.1-1.4 f8f1daa806004e89']})
self.check_task(task)
resp = self.put_task("/api/publish/" + prefix + "/wheezy", task = self.put_task("/api/publish/" + prefix + "/wheezy",
json={ json={
"Signing": DefaultSigningOptions, "Signing": DefaultSigningOptions,
"SkipCleanup": True, "SkipCleanup": True,
}) })
self.check_equal(resp.json()['State'], TASK_SUCCEEDED) self.check_task(task)
repo_expected = { repo_expected = {
'AcquireByHash': False, 'AcquireByHash': False,
'Architectures': ['i386', 'source'], 'Architectures': ['i386', 'source'],
@@ -471,7 +480,8 @@ class PublishUpdateSkipCleanupAPITestRepo(APITest):
self.check_exists("public/" + prefix + self.check_exists("public/" + prefix +
"/pool/main/p/pyspi/pyspi-0.6.1-1.3.stripped.dsc") "/pool/main/p/pyspi/pyspi-0.6.1-1.3.stripped.dsc")
self.check_equal(self.delete_task("/api/publish/" + prefix + "/wheezy", params={"SkipCleanup": "1"}).json()['State'], TASK_SUCCEEDED) task = self.delete_task("/api/publish/" + prefix + "/wheezy", params={"SkipCleanup": "1"})
self.check_task(task)
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/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/p/pyspi/pyspi-0.6.1-1.3.stripped.dsc")
@@ -493,13 +503,15 @@ class PublishSwitchAPITestRepo(APITest):
"pyspi_0.6.1-1.3.dsc", "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.diff.gz", "pyspi_0.6.1.orig.tar.gz",
"pyspi-0.6.1-1.3.stripped.dsc").status_code, 200) "pyspi-0.6.1-1.3.stripped.dsc").status_code, 200)
self.check_equal(self.post_task("/api/repos/" + repo_name + "/file/" + d).json()['State'], TASK_SUCCEEDED) task = self.post_task("/api/repos/" + repo_name + "/file/" + d)
self.check_task(task)
snapshot1_name = self.random_name() snapshot1_name = self.random_name()
self.check_equal(self.post_task("/api/repos/" + repo_name + '/snapshots', json={'Name': snapshot1_name}).json()['State'], TASK_SUCCEEDED) task = self.post_task("/api/repos/" + repo_name + '/snapshots', json={'Name': snapshot1_name})
self.check_task(task)
prefix = self.random_name() prefix = self.random_name()
resp = self.post_task( task = self.post_task(
"/api/publish/" + prefix, "/api/publish/" + prefix,
json={ json={
"Architectures": ["i386", "source"], "Architectures": ["i386", "source"],
@@ -507,8 +519,8 @@ class PublishSwitchAPITestRepo(APITest):
"Sources": [{"Name": snapshot1_name}], "Sources": [{"Name": snapshot1_name}],
"Signing": DefaultSigningOptions, "Signing": DefaultSigningOptions,
}) })
self.check_task(task)
self.check_equal(resp.json()['State'], TASK_SUCCEEDED)
repo_expected = { repo_expected = {
'AcquireByHash': False, 'AcquireByHash': False,
'Architectures': ['i386', 'source'], 'Architectures': ['i386', 'source'],
@@ -537,22 +549,25 @@ class PublishSwitchAPITestRepo(APITest):
d = self.random_name() d = self.random_name()
self.check_equal(self.upload("/api/files/" + d, 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_task("/api/repos/" + repo_name + "/file/" + d).json()['State'], TASK_SUCCEEDED) task = self.post_task("/api/repos/" + repo_name + "/file/" + d)
self.check_task(task)
self.check_equal(self.delete_task("/api/repos/" + repo_name + "/packages/", task = self.delete_task("/api/repos/" + repo_name + "/packages/",
json={"PackageRefs": ['Psource pyspi 0.6.1-1.4 f8f1daa806004e89']}).json()['State'], TASK_SUCCEEDED) json={"PackageRefs": ['Psource pyspi 0.6.1-1.4 f8f1daa806004e89']})
self.check_task(task)
snapshot2_name = self.random_name() snapshot2_name = self.random_name()
self.check_equal(self.post_task("/api/repos/" + repo_name + '/snapshots', json={'Name': snapshot2_name}).json()['State'], TASK_SUCCEEDED) task = self.post_task("/api/repos/" + repo_name + '/snapshots', json={'Name': snapshot2_name})
self.check_task(task)
resp = self.put_task( task = self.put_task(
"/api/publish/" + prefix + "/wheezy", "/api/publish/" + prefix + "/wheezy",
json={ json={
"Snapshots": [{"Component": "main", "Name": snapshot2_name}], "Snapshots": [{"Component": "main", "Name": snapshot2_name}],
"Signing": DefaultSigningOptions, "Signing": DefaultSigningOptions,
"SkipContents": True, "SkipContents": True,
}) })
self.check_equal(resp.json()['State'], TASK_SUCCEEDED) self.check_task(task)
repo_expected = { repo_expected = {
'AcquireByHash': False, 'AcquireByHash': False,
'Architectures': ['i386', 'source'], 'Architectures': ['i386', 'source'],
@@ -579,7 +594,8 @@ class PublishSwitchAPITestRepo(APITest):
self.check_not_exists( self.check_not_exists(
"public/" + prefix + "/pool/main/p/pyspi/pyspi-0.6.1-1.3.stripped.dsc") "public/" + prefix + "/pool/main/p/pyspi/pyspi-0.6.1-1.3.stripped.dsc")
self.check_equal(self.delete_task("/api/publish/" + prefix + "/wheezy").json()['State'], TASK_SUCCEEDED) task = self.delete_task("/api/publish/" + prefix + "/wheezy")
self.check_task(task)
self.check_not_exists("public/" + prefix + "dists/") self.check_not_exists("public/" + prefix + "dists/")
@@ -600,13 +616,15 @@ class PublishSwitchAPISkipCleanupTestRepo(APITest):
"pyspi_0.6.1-1.3.dsc", "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.diff.gz", "pyspi_0.6.1.orig.tar.gz",
"pyspi-0.6.1-1.3.stripped.dsc").status_code, 200) "pyspi-0.6.1-1.3.stripped.dsc").status_code, 200)
self.check_equal(self.post_task("/api/repos/" + repo_name + "/file/" + d).json()['State'], TASK_SUCCEEDED) task = self.post_task("/api/repos/" + repo_name + "/file/" + d)
self.check_task(task)
snapshot1_name = self.random_name() snapshot1_name = self.random_name()
self.check_equal(self.post_task("/api/repos/" + repo_name + '/snapshots', json={'Name': snapshot1_name}).json()['State'], TASK_SUCCEEDED) task = self.post_task("/api/repos/" + repo_name + '/snapshots', json={'Name': snapshot1_name})
self.check_task(task)
prefix = self.random_name() prefix = self.random_name()
resp = self.post_task("/api/publish/" + prefix, task = self.post_task("/api/publish/" + prefix,
json={ json={
"Architectures": ["i386", "source"], "Architectures": ["i386", "source"],
"SourceKind": "snapshot", "SourceKind": "snapshot",
@@ -614,7 +632,7 @@ class PublishSwitchAPISkipCleanupTestRepo(APITest):
"Signing": DefaultSigningOptions, "Signing": DefaultSigningOptions,
}) })
self.check_equal(resp.json()['State'], TASK_SUCCEEDED) self.check_task(task)
repo_expected = { repo_expected = {
'AcquireByHash': False, 'AcquireByHash': False,
'Architectures': ['i386', 'source'], 'Architectures': ['i386', 'source'],
@@ -642,7 +660,7 @@ class PublishSwitchAPISkipCleanupTestRepo(APITest):
# Publish two snapshots, so that deleting one while skipping cleanup will # Publish two snapshots, so that deleting one while skipping cleanup will
# not delete the whole prefix. # not delete the whole prefix.
resp = self.post_task("/api/publish/" + prefix, task = self.post_task("/api/publish/" + prefix,
json={ json={
"Architectures": ["i386", "source"], "Architectures": ["i386", "source"],
"Distribution": "otherdist", "Distribution": "otherdist",
@@ -651,7 +669,7 @@ class PublishSwitchAPISkipCleanupTestRepo(APITest):
"Signing": DefaultSigningOptions, "Signing": DefaultSigningOptions,
}) })
self.check_equal(resp.json()['State'], TASK_SUCCEEDED) self.check_task(task)
repo_expected = { repo_expected = {
'AcquireByHash': False, 'AcquireByHash': False,
'Architectures': ['i386', 'source'], 'Architectures': ['i386', 'source'],
@@ -675,22 +693,25 @@ class PublishSwitchAPISkipCleanupTestRepo(APITest):
d = self.random_name() d = self.random_name()
self.check_equal(self.upload("/api/files/" + d, 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_task("/api/repos/" + repo_name + "/file/" + d).json()['State'], TASK_SUCCEEDED) task = self.post_task("/api/repos/" + repo_name + "/file/" + d)
self.check_task(task)
self.check_equal(self.delete_task("/api/repos/" + repo_name + "/packages/", task = self.delete_task("/api/repos/" + repo_name + "/packages/",
json={"PackageRefs": ['Psource pyspi 0.6.1-1.4 f8f1daa806004e89']}).json()['State'], TASK_SUCCEEDED) json={"PackageRefs": ['Psource pyspi 0.6.1-1.4 f8f1daa806004e89']})
self.check_task(task)
snapshot2_name = self.random_name() snapshot2_name = self.random_name()
self.check_equal(self.post_task("/api/repos/" + repo_name + '/snapshots', json={'Name': snapshot2_name}).json()['State'], TASK_SUCCEEDED) task = self.post_task("/api/repos/" + repo_name + '/snapshots', json={'Name': snapshot2_name})
self.check_task(task)
resp = self.put_task("/api/publish/" + prefix + "/wheezy", task = self.put_task("/api/publish/" + prefix + "/wheezy",
json={ json={
"Snapshots": [{"Component": "main", "Name": snapshot2_name}], "Snapshots": [{"Component": "main", "Name": snapshot2_name}],
"Signing": DefaultSigningOptions, "Signing": DefaultSigningOptions,
"SkipCleanup": True, "SkipCleanup": True,
"SkipContents": True, "SkipContents": True,
}) })
self.check_equal(resp.json()['State'], TASK_SUCCEEDED) self.check_task(task)
repo_expected = { repo_expected = {
'AcquireByHash': False, 'AcquireByHash': False,
'Architectures': ['i386', 'source'], 'Architectures': ['i386', 'source'],
@@ -715,7 +736,8 @@ class PublishSwitchAPISkipCleanupTestRepo(APITest):
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/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/p/pyspi/pyspi-0.6.1-1.3.stripped.dsc")
self.check_equal(self.delete_task("/api/publish/" + prefix + "/wheezy", params={"SkipCleanup": "1"}).json()['State'], TASK_SUCCEEDED) task = self.delete_task("/api/publish/" + prefix + "/wheezy", params={"SkipCleanup": "1"})
self.check_task(task)
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/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/p/pyspi/pyspi-0.6.1-1.3.stripped.dsc")
system/t12_api/repos.py
+63 -41
View File
@@ -1,4 +1,4 @@
from api_lib import TASK_FAILED, TASK_SUCCEEDED, APITest from api_lib import APITest
from .publish import DefaultSigningOptions from .publish import DefaultSigningOptions
@@ -41,7 +41,8 @@ class ReposAPITestCreateIndexDelete(APITest):
names = [repo["Name"] for repo in repos] names = [repo["Name"] for repo in repos]
assert repo_name in names assert repo_name in names
self.check_equal(self.delete_task("/api/repos/" + repo_name).json()['State'], TASK_SUCCEEDED) resp = self.delete_task("/api/repos/" + repo_name)
self.check_task(resp)
self.check_equal(self.delete("/api/repos/" + repo_name).status_code, 404) self.check_equal(self.delete("/api/repos/" + repo_name).status_code, 404)
self.check_equal(self.get("/api/repos/" + repo_name).status_code, 404) self.check_equal(self.get("/api/repos/" + repo_name).status_code, 404)
@@ -61,25 +62,32 @@ class ReposAPITestCreateIndexDelete(APITest):
"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) "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_task("/api/repos/" + repo_name + "/file/" + d) resp = self.post_task("/api/repos/" + repo_name + "/file/" + d)
self.check_equal(resp.json()['State'], TASK_SUCCEEDED) self.check_task(resp)
self.check_equal(self.post_task("/api/repos/" + repo_name + "/snapshots", json={"Name": repo_name}).json()['State'], TASK_SUCCEEDED) task = self.post_task("/api/repos/" + repo_name + "/snapshots", json={"Name": repo_name})
self.check_task(task)
self.check_equal(self.post_task("/api/publish", resp = self.post_task("/api/publish",
json={ json={
"SourceKind": "local", "SourceKind": "local",
"Sources": [{"Name": repo_name}], "Sources": [{"Name": repo_name}],
"Signing": DefaultSigningOptions, "Signing": DefaultSigningOptions,
}).json()['State'], TASK_SUCCEEDED) })
self.check_task(resp)
# repo is not deletable while it is published # repo is not deletable while it is published
self.check_equal(self.delete_task("/api/repos/" + repo_name).json()['State'], TASK_FAILED) task = self.delete_task("/api/repos/" + repo_name)
self.check_equal(self.delete_task("/api/repos/" + repo_name, params={"force": "1"}).json()['State'], TASK_FAILED) self.check_task_fail(task)
task = self.delete_task("/api/repos/" + repo_name, params={"force": "1"})
self.check_task_fail(task)
# drop published # drop published
self.check_equal(self.delete_task("/api/publish//" + distribution).json()['State'], TASK_SUCCEEDED) task = self.delete_task("/api/publish//" + distribution)
self.check_equal(self.delete_task("/api/repos/" + repo_name).json()['State'], TASK_FAILED) self.check_task(task)
self.check_equal(self.delete_task("/api/repos/" + repo_name, params={"force": "1"}).json()['State'], TASK_SUCCEEDED) task = self.delete_task("/api/repos/" + repo_name)
self.check_task_fail(task)
task = self.delete_task("/api/repos/" + repo_name, params={"force": "1"})
self.check_task(task)
self.check_equal(self.get("/api/repos/" + repo_name).status_code, 404) self.check_equal(self.get("/api/repos/" + repo_name).status_code, 404)
@@ -97,7 +105,7 @@ class ReposAPITestAdd(APITest):
"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) "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_task("/api/repos/" + repo_name + "/file/" + d) resp = self.post_task("/api/repos/" + repo_name + "/file/" + d)
self.check_equal(resp.json()['State'], TASK_SUCCEEDED) self.check_task(resp)
resp = self.get("/api/tasks/" + str(resp.json()['ID']) + "/output") resp = self.get("/api/tasks/" + str(resp.json()['ID']) + "/output")
self.check_equal(resp.status_code, 200) self.check_equal(resp.status_code, 200)
@@ -125,7 +133,8 @@ class ReposAPITestAddNotFullRemove(APITest):
self.check_equal(self.upload("/api/files/" + d, 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) "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_task("/api/repos/" + repo_name + "/file/" + d).json()['State'], TASK_SUCCEEDED) task = self.post_task("/api/repos/" + repo_name + "/file/" + d)
self.check_task(task)
self.check_equal(self.get("/api/repos/" + repo_name + "/packages").json(), ['Psource pyspi 0.6.1-1.3 3a8b37cbd9a3559e']) 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") self.check_exists("upload/" + d + "/aptly.pub")
@@ -145,7 +154,8 @@ class ReposAPITestAddNoRemove(APITest):
self.check_equal(self.upload("/api/files/" + d, 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) "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_task("/api/repos/" + repo_name + "/file/" + d, params={"noRemove": 1}).json()['State'], TASK_SUCCEEDED) task = self.post_task("/api/repos/" + repo_name + "/file/" + d, params={"noRemove": 1})
self.check_task(task)
self.check_equal(self.get("/api/repos/" + repo_name + "/packages").json(), ['Psource pyspi 0.6.1-1.3 3a8b37cbd9a3559e']) 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") self.check_exists("upload/" + d + "/pyspi_0.6.1-1.3.dsc")
@@ -165,7 +175,7 @@ class ReposAPITestAddFile(APITest):
"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)
resp = self.post_task("/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'], TASK_SUCCEEDED) self.check_task(resp)
resp = self.get("/api/tasks/" + str(resp.json()['ID']) + "/output") resp = self.get("/api/tasks/" + str(resp.json()['ID']) + "/output")
self.check_equal(resp.status_code, 200) self.check_equal(resp.status_code, 200)
@@ -197,7 +207,7 @@ class ReposAPITestInclude(APITest):
self.check_equal(resp.status_code, 200) self.check_equal(resp.status_code, 200)
resp = self.post_task("/api/repos/" + repo_name + "/include/" + d, params={"ignoreSignature": 1}) resp = self.post_task("/api/repos/" + repo_name + "/include/" + d, params={"ignoreSignature": 1})
self.check_equal(resp.json()['State'], TASK_SUCCEEDED) self.check_task(resp)
resp = self.get("/api/tasks/" + str(resp.json()['ID']) + "/output") resp = self.get("/api/tasks/" + str(resp.json()['ID']) + "/output")
self.check_equal(resp.status_code, 200) self.check_equal(resp.status_code, 200)
@@ -225,7 +235,8 @@ class ReposAPITestShowQuery(APITest):
"libboost-program-options-dev_1.49.0.1_i386.deb", "pyspi_0.6.1-1.3.dsc", "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.diff.gz", "pyspi_0.6.1.orig.tar.gz",
"pyspi-0.6.1-1.3.stripped.dsc").status_code, 200) "pyspi-0.6.1-1.3.stripped.dsc").status_code, 200)
self.check_equal(self.post_task("/api/repos/" + repo_name + "/file/" + d).json()['State'], TASK_SUCCEEDED) task = self.post_task("/api/repos/" + repo_name + "/file/" + d)
self.check_task(task)
self.check_equal(sorted(self.get("/api/repos/" + repo_name + "/packages", params={"q": "pyspi"}).json()), 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']) ['Psource pyspi 0.6.1-1.3 3a8b37cbd9a3559e', 'Psource pyspi 0.6.1-1.4 f8f1daa806004e89'])
@@ -258,13 +269,15 @@ class ReposAPITestAddMultiple(APITest):
"pyspi_0.6.1-1.3.diff.gz", "pyspi_0.6.1.orig.tar.gz", "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) "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.dsc", task = self.post_task("/api/repos/" + repo_name + "/file/" + d + "/pyspi_0.6.1-1.3.dsc",
params={"noRemove": 1}).json()['State'], TASK_SUCCEEDED) params={"noRemove": 1})
self.check_task(task)
self.check_equal(sorted(self.get("/api/repos/" + repo_name + "/packages").json()), 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.3 3a8b37cbd9a3559e'])
self.check_equal(self.post_task("/api/repos/" + repo_name + "/file/" + d + "/pyspi-0.6.1-1.3.stripped.dsc").json()['State'], TASK_SUCCEEDED) task = self.post_task("/api/repos/" + repo_name + "/file/" + d + "/pyspi-0.6.1-1.3.stripped.dsc")
self.check_task(task)
self.check_equal(sorted(self.get("/api/repos/" + repo_name + "/packages").json()), 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']) ['Psource pyspi 0.6.1-1.3 3a8b37cbd9a3559e', 'Psource pyspi 0.6.1-1.4 f8f1daa806004e89'])
@@ -286,34 +299,39 @@ class ReposAPITestPackagesAddDelete(APITest):
"pyspi_0.6.1-1.3.diff.gz", "pyspi_0.6.1.orig.tar.gz", "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) "pyspi-0.6.1-1.3.stripped.dsc").status_code, 200)
self.check_equal(self.post_task("/api/repos/" + repo_name + "/file/" + d).json()['State'], TASK_SUCCEEDED) task = self.post_task("/api/repos/" + repo_name + "/file/" + d)
self.check_task(task)
self.check_equal(sorted(self.get("/api/repos/" + repo_name + "/packages").json()), self.check_equal(sorted(self.get("/api/repos/" + repo_name + "/packages").json()),
['Pi386 libboost-program-options-dev 1.49.0.1 918d2f433384e378', ['Pi386 libboost-program-options-dev 1.49.0.1 918d2f433384e378',
'Psource pyspi 0.6.1-1.3 3a8b37cbd9a3559e', 'Psource pyspi 0.6.1-1.3 3a8b37cbd9a3559e',
'Psource pyspi 0.6.1-1.4 f8f1daa806004e89']) 'Psource pyspi 0.6.1-1.4 f8f1daa806004e89'])
self.check_equal(self.post_task("/api/repos/" + repo_name + "/packages/", task = self.post_task("/api/repos/" + repo_name + "/packages/",
json={"PackageRefs": ['Psource pyspi 0.6.1-1.4 f8f1daa806004e89']}).json()['State'], TASK_SUCCEEDED) json={"PackageRefs": ['Psource pyspi 0.6.1-1.4 f8f1daa806004e89']})
self.check_task(task)
self.check_equal(sorted(self.get("/api/repos/" + repo_name + "/packages").json()), self.check_equal(sorted(self.get("/api/repos/" + repo_name + "/packages").json()),
['Pi386 libboost-program-options-dev 1.49.0.1 918d2f433384e378', ['Pi386 libboost-program-options-dev 1.49.0.1 918d2f433384e378',
'Psource pyspi 0.6.1-1.3 3a8b37cbd9a3559e', 'Psource pyspi 0.6.1-1.3 3a8b37cbd9a3559e',
'Psource pyspi 0.6.1-1.4 f8f1daa806004e89']) 'Psource pyspi 0.6.1-1.4 f8f1daa806004e89'])
self.check_equal(self.post_task("/api/repos/" + repo_name + "/packages/", task = self.post_task("/api/repos/" + repo_name + "/packages/",
json={"PackageRefs": ['Psource pyspi 0.6.1-1.4 f8f1daa806004e89', json={"PackageRefs": ['Psource pyspi 0.6.1-1.4 f8f1daa806004e89',
'Psource no-such-package 0.6.1-1.4 f8f1daa806004e89']}).json()['State'], TASK_FAILED) 'Psource no-such-package 0.6.1-1.4 f8f1daa806004e89']})
self.check_task_fail(task)
self.check_equal(self.delete_task("/api/repos/" + repo_name + "/packages/", task = self.delete_task("/api/repos/" + repo_name + "/packages/",
json={"PackageRefs": ['Psource pyspi 0.6.1-1.4 f8f1daa806004e89']}).json()['State'], TASK_SUCCEEDED) json={"PackageRefs": ['Psource pyspi 0.6.1-1.4 f8f1daa806004e89']})
self.check_task(task)
self.check_equal(sorted(self.get("/api/repos/" + repo_name + "/packages").json()), self.check_equal(sorted(self.get("/api/repos/" + repo_name + "/packages").json()),
['Pi386 libboost-program-options-dev 1.49.0.1 918d2f433384e378', ['Pi386 libboost-program-options-dev 1.49.0.1 918d2f433384e378',
'Psource pyspi 0.6.1-1.3 3a8b37cbd9a3559e']) 'Psource pyspi 0.6.1-1.3 3a8b37cbd9a3559e'])
self.check_equal(self.post_task("/api/repos/" + repo_name + "/packages/", task = self.post_task("/api/repos/" + repo_name + "/packages/",
json={"PackageRefs": ['Psource pyspi 0.6.1-1.4 f8f1daa806004e89']}).json()['State'], TASK_SUCCEEDED) json={"PackageRefs": ['Psource pyspi 0.6.1-1.4 f8f1daa806004e89']})
self.check_task(task)
self.check_equal(sorted(self.get("/api/repos/" + repo_name + "/packages").json()), self.check_equal(sorted(self.get("/api/repos/" + repo_name + "/packages").json()),
['Pi386 libboost-program-options-dev 1.49.0.1 918d2f433384e378', ['Pi386 libboost-program-options-dev 1.49.0.1 918d2f433384e378',
@@ -324,9 +342,10 @@ 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", json={"Name": repo_name2, "Comment": "fun repo"}).status_code, 201)
self.check_equal(self.post_task("/api/repos/" + repo_name2 + "/packages/", task = self.post_task("/api/repos/" + repo_name2 + "/packages/",
json={"PackageRefs": ['Psource pyspi 0.6.1-1.4 f8f1daa806004e89', json={"PackageRefs": ['Psource pyspi 0.6.1-1.4 f8f1daa806004e89',
'Pi386 libboost-program-options-dev 1.49.0.1 918d2f433384e378']}).json()['State'], TASK_SUCCEEDED) 'Pi386 libboost-program-options-dev 1.49.0.1 918d2f433384e378']})
self.check_task(task)
self.check_equal(sorted(self.get("/api/repos/" + repo_name2 + "/packages").json()), self.check_equal(sorted(self.get("/api/repos/" + repo_name2 + "/packages").json()),
['Pi386 libboost-program-options-dev 1.49.0.1 918d2f433384e378', ['Pi386 libboost-program-options-dev 1.49.0.1 918d2f433384e378',
@@ -349,7 +368,7 @@ class ReposAPITestShowMaxVersion(APITest):
).status_code, 200) ).status_code, 200)
resp = self.post_task("/api/repos/" + repo_name + "/file/" + d) resp = self.post_task("/api/repos/" + repo_name + "/file/" + d)
self.check_equal(resp.json()['State'], TASK_SUCCEEDED) self.check_task(resp)
resp = self.get("/api/tasks/" + str(resp.json()['ID']) + "/output") resp = self.get("/api/tasks/" + str(resp.json()['ID']) + "/output")
self.check_equal(resp.status_code, 200) self.check_equal(resp.status_code, 200)
@@ -377,7 +396,7 @@ class ReposAPITestCopyPackage(APITest):
d = self.random_name() d = self.random_name()
self.check_equal(self.upload(f"/api/files/{d}", f"{pkg_name}.deb").status_code, 200) self.check_equal(self.upload(f"/api/files/{d}", f"{pkg_name}.deb").status_code, 200)
resp = self.post_task(f"/api/repos/{repo1_name}/file/{d}") resp = self.post_task(f"/api/repos/{repo1_name}/file/{d}")
self.check_equal(resp.json()['State'], TASK_SUCCEEDED) self.check_task(resp)
# Creating target repo # Creating target repo
repo2_name = self.random_name() repo2_name = self.random_name()
@@ -385,8 +404,11 @@ class ReposAPITestCopyPackage(APITest):
# Copy the package # Copy the package
resp = self.post_task(f"/api/repos/{repo2_name}/copy/{repo1_name}/{pkg_name}") resp = self.post_task(f"/api/repos/{repo2_name}/copy/{repo1_name}/{pkg_name}")
self.check_equal(resp.status_code, 200) self.check_task(resp)
self.check_equal(resp.json()['State'], TASK_SUCCEEDED)
# Test non-existing package
resp = self.post_task(f"/api/repos/{repo2_name}/copy/{repo1_name}/la<la>la")
self.check_task_fail(resp)
self.check_equal(self.get(f"/api/repos/{repo2_name}/packages").json(), self.check_equal(self.get(f"/api/repos/{repo2_name}/packages").json(),
['Pi386 libboost-program-options-dev 1.49.0.1 918d2f433384e378']) ['Pi386 libboost-program-options-dev 1.49.0.1 918d2f433384e378'])
system/t12_api/snapshots.py
+45 -39
View File
@@ -1,4 +1,4 @@
from api_lib import TASK_FAILED, TASK_SUCCEEDED, APITest from api_lib import TASK_FAILED, APITest
from .publish import DefaultSigningOptions from .publish import DefaultSigningOptions
@@ -13,8 +13,8 @@ class SnapshotsAPITestCreateShowEmpty(APITest):
'Name': snapshot_name} 'Name': snapshot_name}
# create empty snapshot # create empty snapshot
resp = self.post_task("/api/snapshots", json=snapshot_desc) task = self.post_task("/api/snapshots", json=snapshot_desc)
self.check_equal(resp.json()['State'], TASK_SUCCEEDED) self.check_task(task)
self.check_subset(snapshot_desc, self.get("/api/snapshots/" + snapshot_name).json()) self.check_subset(snapshot_desc, self.get("/api/snapshots/" + snapshot_name).json())
self.check_equal(self.get("/api/snapshots/" + snapshot_name).status_code, 200) self.check_equal(self.get("/api/snapshots/" + snapshot_name).status_code, 200)
@@ -47,8 +47,8 @@ class SnapshotsAPITestCreateFromRefs(APITest):
# create empty snapshot # create empty snapshot
empty_snapshot_name = self.random_name() empty_snapshot_name = self.random_name()
resp = self.post_task("/api/snapshots", json={"Name": empty_snapshot_name}) task = self.post_task("/api/snapshots", json={"Name": empty_snapshot_name})
self.check_equal(resp.json()['State'], TASK_SUCCEEDED) self.check_task(task)
self.check_equal( self.check_equal(
self.get("/api/snapshots/" + empty_snapshot_name).json()['Description'], "Created as empty" self.get("/api/snapshots/" + empty_snapshot_name).json()['Description'], "Created as empty"
) )
@@ -59,14 +59,15 @@ class SnapshotsAPITestCreateFromRefs(APITest):
d = self.random_name() d = self.random_name()
self.check_equal(self.upload("/api/files/" + d, 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_task("/api/repos/" + repo_name + "/file/" + d).json()['State'], TASK_SUCCEEDED) task = self.post_task("/api/repos/" + repo_name + "/file/" + d)
self.check_task(task)
# create snapshot with empty snapshot as source and package # create snapshot with empty snapshot as source and package
snapshot = snapshot_desc.copy() snapshot = snapshot_desc.copy()
snapshot['PackageRefs'] = ["Pi386 libboost-program-options-dev 1.49.0.1 918d2f433384e378"] snapshot['PackageRefs'] = ["Pi386 libboost-program-options-dev 1.49.0.1 918d2f433384e378"]
snapshot['SourceSnapshots'] = [empty_snapshot_name] snapshot['SourceSnapshots'] = [empty_snapshot_name]
resp = self.post_task("/api/snapshots", json=snapshot) task = self.post_task("/api/snapshots", json=snapshot)
self.check_equal(resp.json()['State'], TASK_SUCCEEDED) self.check_task(task)
snapshot.pop('SourceSnapshots') snapshot.pop('SourceSnapshots')
snapshot.pop('PackageRefs') snapshot.pop('PackageRefs')
resp = self.get("/api/snapshots/" + snapshot_name) resp = self.get("/api/snapshots/" + snapshot_name)
@@ -100,8 +101,8 @@ class SnapshotsAPITestCreateFromRepo(APITest):
snapshot_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)
resp = self.post_task("/api/repos/" + repo_name + '/snapshots', json={'Name': snapshot_name}) task = self.post_task("/api/repos/" + repo_name + '/snapshots', json={'Name': snapshot_name})
self.check_equal(resp.json()['State'], TASK_SUCCEEDED) self.check_task(task)
self.check_equal([], self.check_equal([],
self.get("/api/snapshots/" + snapshot_name + "/packages", params={"format": "details"}).json()) self.get("/api/snapshots/" + snapshot_name + "/packages", params={"format": "details"}).json())
@@ -110,10 +111,11 @@ class SnapshotsAPITestCreateFromRepo(APITest):
self.check_equal(self.upload("/api/files/" + d, 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_task("/api/repos/" + repo_name + "/file/" + d).json()['State'], TASK_SUCCEEDED) task = self.post_task("/api/repos/" + repo_name + "/file/" + d)
self.check_task(task)
resp = self.post_task("/api/repos/" + repo_name + '/snapshots', json={'Name': snapshot_name}) task = self.post_task("/api/repos/" + repo_name + '/snapshots', json={'Name': snapshot_name})
self.check_equal(resp.json()['State'], TASK_SUCCEEDED) self.check_task(task)
self.check_equal(self.get("/api/snapshots/" + snapshot_name).status_code, 200) self.check_equal(self.get("/api/snapshots/" + snapshot_name).status_code, 200)
self.check_subset({'Architecture': 'i386', self.check_subset({'Architecture': 'i386',
@@ -143,13 +145,13 @@ class SnapshotsAPITestCreateUpdate(APITest):
snapshot_desc = {'Description': 'fun snapshot', snapshot_desc = {'Description': 'fun snapshot',
'Name': snapshot_name} 'Name': snapshot_name}
resp = self.post_task("/api/snapshots", json=snapshot_desc) task = self.post_task("/api/snapshots", json=snapshot_desc)
self.check_equal(resp.json()['State'], TASK_SUCCEEDED) self.check_task(task)
new_snapshot_name = self.random_name() new_snapshot_name = self.random_name()
resp = self.put_task("/api/snapshots/" + snapshot_name, json={'Name': new_snapshot_name, task = self.put_task("/api/snapshots/" + snapshot_name, json={'Name': new_snapshot_name,
'Description': 'New description'}) 'Description': 'New description'})
self.check_equal(resp.json()['State'], TASK_SUCCEEDED) self.check_task(task)
resp = self.get("/api/snapshots/" + new_snapshot_name) resp = self.get("/api/snapshots/" + new_snapshot_name)
self.check_equal(resp.status_code, 200) self.check_equal(resp.status_code, 200)
@@ -176,16 +178,18 @@ class SnapshotsAPITestCreateDelete(APITest):
'Name': snapshot_name} 'Name': snapshot_name}
# deleting unreferenced snapshot # deleting unreferenced snapshot
resp = self.post_task("/api/snapshots", json=snapshot_desc) task = self.post_task("/api/snapshots", json=snapshot_desc)
self.check_equal(resp.json()['State'], TASK_SUCCEEDED) self.check_task(task)
self.check_equal(self.delete_task("/api/snapshots/" + snapshot_name).json()['State'], TASK_SUCCEEDED) task = self.delete_task("/api/snapshots/" + snapshot_name)
self.check_task(task)
self.check_equal(self.get("/api/snapshots/" + snapshot_name).status_code, 404) self.check_equal(self.get("/api/snapshots/" + snapshot_name).status_code, 404)
# deleting referenced snapshot # deleting referenced snapshot
snap1, snap2 = self.random_name(), self.random_name() snap1, snap2 = self.random_name(), self.random_name()
self.check_equal(self.post_task("/api/snapshots", json={"Name": snap1}).json()['State'], TASK_SUCCEEDED) task = self.post_task("/api/snapshots", json={"Name": snap1})
self.check_task(task)
self.check_equal( self.check_equal(
self.post_task( self.post_task(
"/api/snapshots", json={"Name": snap2, "SourceSnapshots": [snap1]} "/api/snapshots", json={"Name": snap2, "SourceSnapshots": [snap1]}
@@ -194,11 +198,12 @@ class SnapshotsAPITestCreateDelete(APITest):
self.check_equal(self.delete_task("/api/snapshots/" + snap1).json()['State'], TASK_FAILED) self.check_equal(self.delete_task("/api/snapshots/" + snap1).json()['State'], TASK_FAILED)
self.check_equal(self.get("/api/snapshots/" + snap1).status_code, 200) self.check_equal(self.get("/api/snapshots/" + snap1).status_code, 200)
self.check_equal(self.delete_task("/api/snapshots/" + snap1, params={"force": "1"}).json()['State'], TASK_SUCCEEDED) 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) self.check_equal(self.get("/api/snapshots/" + snap1).status_code, 404)
# deleting published snapshot # deleting published snapshot
resp = self.post_task( task = self.post_task(
"/api/publish", "/api/publish",
json={ json={
"SourceKind": "snapshot", "SourceKind": "snapshot",
@@ -208,7 +213,7 @@ class SnapshotsAPITestCreateDelete(APITest):
"Signing": DefaultSigningOptions, "Signing": DefaultSigningOptions,
} }
) )
self.check_equal(resp.json()['State'], TASK_SUCCEEDED) self.check_task(task)
self.check_equal(self.delete_task("/api/snapshots/" + snap2).json()['State'], TASK_FAILED) self.check_equal(self.delete_task("/api/snapshots/" + snap2).json()['State'], TASK_FAILED)
self.check_equal(self.delete_task("/api/snapshots/" + snap2, params={"force": "1"}).json()['State'], TASK_FAILED) self.check_equal(self.delete_task("/api/snapshots/" + snap2, params={"force": "1"}).json()['State'], TASK_FAILED)
@@ -228,10 +233,11 @@ class SnapshotsAPITestSearch(APITest):
self.check_equal(self.upload("/api/files/" + d, 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_task("/api/repos/" + repo_name + "/file/" + d).json()['State'], TASK_SUCCEEDED) task = self.post_task("/api/repos/" + repo_name + "/file/" + d)
self.check_task(task)
resp = self.post_task("/api/repos/" + repo_name + '/snapshots', json={'Name': snapshot_name}) task = self.post_task("/api/repos/" + repo_name + '/snapshots', json={'Name': snapshot_name})
self.check_equal(resp.json()['State'], TASK_SUCCEEDED) self.check_task(task)
resp = self.get("/api/snapshots/" + snapshot_name + "/packages", resp = self.get("/api/snapshots/" + snapshot_name + "/packages",
params={"q": "libboost-program-options-dev", "format": "details"}) params={"q": "libboost-program-options-dev", "format": "details"})
@@ -262,13 +268,14 @@ class SnapshotsAPITestDiff(APITest):
self.check_equal(self.upload("/api/files/" + d, 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_task("/api/repos/" + repos[-1] + "/file/" + d).json()['State'], TASK_SUCCEEDED) task = self.post_task("/api/repos/" + repos[-1] + "/file/" + d)
self.check_task(task)
resp = self.post_task("/api/repos/" + repos[-1] + '/snapshots', json={'Name': snapshots[0]}) task = self.post_task("/api/repos/" + repos[-1] + '/snapshots', json={'Name': snapshots[0]})
self.check_equal(resp.json()['State'], TASK_SUCCEEDED) self.check_task(task)
resp = self.post_task("/api/snapshots", json={'Name': snapshots[1]}) task = self.post_task("/api/snapshots", json={'Name': snapshots[1]})
self.check_equal(resp.json()['State'], TASK_SUCCEEDED) self.check_task(task)
resp = self.get("/api/snapshots/" + snapshots[0] + "/diff/" + snapshots[1]) resp = self.get("/api/snapshots/" + snapshots[0] + "/diff/" + snapshots[1])
self.check_equal(resp.status_code, 200) self.check_equal(resp.status_code, 200)
@@ -302,19 +309,19 @@ class SnapshotsAPITestMerge(APITest):
# create source snapshots # create source snapshots
for source in sources: for source in sources:
resp = self.post_task("/api/snapshots", json=source) task = self.post_task("/api/snapshots", json=source)
self.check_equal(resp.json()["State"], TASK_SUCCEEDED) self.check_task(task)
# create merge snapshot # create merge snapshot
merged_name = self.random_name() merged_name = self.random_name()
resp = self.post_task( task = self.post_task(
"/api/snapshots/merge", "/api/snapshots/merge",
json={ json={
"Destination": merged_name, "Destination": merged_name,
"Sources": [source["Name"] for source in sources], "Sources": [source["Name"] for source in sources],
}, },
) )
self.check_equal(resp.json()["State"], TASK_SUCCEEDED) self.check_task(task)
# check merge snapshot # check merge snapshot
resp = self.get(f"/api/snapshots/{merged_name}") resp = self.get(f"/api/snapshots/{merged_name}")
@@ -329,9 +336,8 @@ class SnapshotsAPITestMerge(APITest):
) )
# remove merge snapshot # remove merge snapshot
self.check_equal( task = self.delete_task(f"/api/snapshots/{merged_name}")
self.delete_task(f"/api/snapshots/{merged_name}").json()["State"], TASK_SUCCEEDED self.check_task(task)
)
# create merge snapshot without sources # create merge snapshot without sources
merged_name = self.random_name() merged_name = self.random_name()
system/t12_api/tasks.py
+5 -7
View File
@@ -1,4 +1,4 @@
from api_lib import TASK_SUCCEEDED, APITest from api_lib import APITest
from .publish import DefaultSigningOptions from .publish import DefaultSigningOptions
@@ -49,9 +49,8 @@ class TaskAPITestParallelTasks(APITest):
def _wait_for_task(self, task_id): def _wait_for_task(self, task_id):
uri = "/api/tasks/%d/wait" % int(task_id) uri = "/api/tasks/%d/wait" % int(task_id)
resp = self.get(uri) task = self.get(uri)
self.check_equal(resp.status_code, 200) self.check_task(task)
self.check_equal(resp.json()['State'], TASK_SUCCEEDED)
def _wait_for_all_tasks(self): def _wait_for_all_tasks(self):
resp = self.get("/api/tasks-wait") resp = self.get("/api/tasks-wait")
@@ -103,6 +102,5 @@ class TaskAPITestParallelTasks(APITest):
self._wait_for_all_tasks() self._wait_for_all_tasks()
for publish_task_id in publish_task_ids: for publish_task_id in publish_task_ids:
resp = self.get("/api/tasks/%d" % publish_task_id) task = self.get("/api/tasks/%d" % publish_task_id)
self.check_equal(resp.status_code, 200) self.check_task(task)
self.check_equal(resp.json()['State'], TASK_SUCCEEDED)
system/t12_api/unix_socket.py
+10 -1
View File
@@ -6,19 +6,28 @@ import urllib.parse
import urllib.request import urllib.request
from lib import BaseTest from lib import BaseTest
from api_lib import AptlyStream
class UnixSocketAPITest(BaseTest): class UnixSocketAPITest(BaseTest):
aptly_server = None aptly_server = None
socket_path = "/tmp/_aptly_test.sock" socket_path = "/tmp/_aptly_test.sock"
base_url = ("unix://%s" % socket_path) base_url = ("unix://%s" % socket_path)
aptly_out = None
debugOutput = True
def prepare(self): def prepare(self):
if self.aptly_server is None: if self.aptly_server is None:
self.aptly_server = self._start_process("aptly api serve -no-lock -listen=%s" % (self.base_url),) UnixSocketAPITest.aptly_out = AptlyStream()
self.aptly_server = self._start_process("aptly api serve -no-lock -listen=%s" % (self.base_url), stdout=UnixSocketAPITest.aptly_out, stderr=UnixSocketAPITest.aptly_out)
time.sleep(1) time.sleep(1)
else:
UnixSocketAPITest.aptly_out.clear()
super(UnixSocketAPITest, self).prepare() super(UnixSocketAPITest, self).prepare()
def debug_output(self):
return UnixSocketAPITest.aptly_out.get_contents()
def shutdown(self): def shutdown(self):
if self.aptly_server is not None: if self.aptly_server is not None:
self.aptly_server.terminate() self.aptly_server.terminate()