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

implement system tests for serving api and published repos simultaneously

Browse Source
This commit is contained in:
Markus Muellner
2023-03-17 17:04:02 +01:00
committed by Benj Fassbind
parent e25ade8af3
commit f74217ed9c
4 changed files with 119 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
system/api_lib.py
+12 -3
View File
@@ -24,6 +24,14 @@ class APITest(BaseTest):
"""
aptly_server = None
base_url = "127.0.0.1:8765"
configOverride = {
"FileSystemPublishEndpoints": {
"apiandserve": {
"rootDir": f"{os.environ['HOME']}/{BaseTest.aptlyDir}/apiandserve",
"linkMethod": "symlink"
}
}
}
def fixture_available(self):
return super(APITest, self).fixture_available() and requests is not None
@@ -32,11 +40,12 @@ class APITest(BaseTest):
if APITest.aptly_server is None:
super(APITest, self).prepare()
APITest.aptly_server = self._start_process("aptly api serve -no-lock -listen=%s" % (self.base_url),)
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}",)
time.sleep(1)
if os.path.exists(os.path.join(os.environ["HOME"], ".aptly", "upload")):
shutil.rmtree(os.path.join(os.environ["HOME"], ".aptly", "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"))
def run(self):
pass
system/lib.py
+19 -17
View File
@@ -128,9 +128,11 @@ class BaseTest(object):
requiresDot = False
sortOutput = False
aptlyDir = ".aptly"
aptlyConfigFile = ".aptly.conf"
expectedCode = 0
configFile = {
"rootDir": "%s/.aptly" % os.environ["HOME"],
"rootDir": f"{os.environ['HOME']}/{aptlyDir}",
"downloadConcurrency": 4,
"downloadSpeedLimit": 0,
"downloadRetries": 5,
@@ -177,10 +179,10 @@ class BaseTest(object):
self.teardown()
def prepare_remove_all(self):
if os.path.exists(os.path.join(os.environ["HOME"], ".aptly")):
shutil.rmtree(os.path.join(os.environ["HOME"], ".aptly"))
if os.path.exists(os.path.join(os.environ["HOME"], ".aptly.conf")):
os.remove(os.path.join(os.environ["HOME"], ".aptly.conf"))
if os.path.exists(os.path.join(os.environ["HOME"], self.aptlyDir)):
shutil.rmtree(os.path.join(os.environ["HOME"], self.aptlyDir))
if os.path.exists(os.path.join(os.environ["HOME"], self.aptlyConfigFile)):
os.remove(os.path.join(os.environ["HOME"], self.aptlyConfigFile))
if os.path.exists(os.path.join(os.environ["HOME"], ".gnupg", "aptlytest.gpg")):
os.remove(os.path.join(
os.environ["HOME"], ".gnupg", "aptlytest.gpg"))
@@ -192,7 +194,7 @@ class BaseTest(object):
elif self.requiresGPG2:
cfg["gpgProvider"] = "gpg2"
cfg.update(**self.configOverride)
f = open(os.path.join(os.environ["HOME"], ".aptly.conf"), "w")
f = open(os.path.join(os.environ["HOME"], self.aptlyConfigFile), "w")
f.write(json.dumps(cfg))
f.close()
@@ -214,18 +216,18 @@ class BaseTest(object):
def prepare_fixture(self):
if self.fixturePool:
os.makedirs(os.path.join(os.environ["HOME"], ".aptly"), 0o755)
os.makedirs(os.path.join(os.environ["HOME"], self.aptlyDir), 0o755)
os.symlink(self.fixturePoolDir, os.path.join(
os.environ["HOME"], ".aptly", "pool"))
os.environ["HOME"], self.aptlyDir, "pool"))
if self.fixturePoolCopy:
os.makedirs(os.path.join(os.environ["HOME"], ".aptly"), 0o755)
os.makedirs(os.path.join(os.environ["HOME"], self.aptlyDir), 0o755)
shutil.copytree(self.fixturePoolDir, os.path.join(
os.environ["HOME"], ".aptly", "pool"), ignore=shutil.ignore_patterns(".git"))
os.environ["HOME"], self.aptlyDir, "pool"), ignore=shutil.ignore_patterns(".git"))
if self.fixtureDB:
shutil.copytree(self.fixtureDBDir, os.path.join(
os.environ["HOME"], ".aptly", "db"))
os.environ["HOME"], self.aptlyDir, "db"))
if self.fixtureWebServer:
self.webServerUrl = self.start_webserver(os.path.join(os.path.dirname(inspect.getsourcefile(self.__class__)),
@@ -260,7 +262,7 @@ class BaseTest(object):
'changes': os.path.join(os.path.dirname(inspect.getsourcefile(BaseTest)), "changes"),
'udebs': os.path.join(os.path.dirname(inspect.getsourcefile(BaseTest)), "udebs"),
'testfiles': os.path.join(os.path.dirname(inspect.getsourcefile(self.__class__)), self.__class__.__name__),
'aptlyroot': os.path.join(os.environ["HOME"], ".aptly"),
'aptlyroot': os.path.join(os.environ["HOME"], self.aptlyDir),
}
if self.fixtureWebServer:
params['url'] = self.webServerUrl
@@ -365,11 +367,11 @@ class BaseTest(object):
raise
def read_file(self, path, mode=''):
with open(os.path.join(os.environ["HOME"], ".aptly", path), "r" + mode) as f:
with open(os.path.join(os.environ["HOME"], self.aptlyDir, path), "r" + mode) as f:
return f.read()
def delete_file(self, path):
os.unlink(os.path.join(os.environ["HOME"], ".aptly", path))
os.unlink(os.path.join(os.environ["HOME"], self.aptlyDir, path))
def check_file_contents(self, path, gold_name, match_prepare=None, mode='', ensure_utf8=True):
contents = self.read_file(path, mode=mode)
@@ -399,15 +401,15 @@ class BaseTest(object):
raise
def check_exists(self, path):
if not os.path.exists(os.path.join(os.environ["HOME"], ".aptly", path)):
if not os.path.exists(os.path.join(os.environ["HOME"], self.aptlyDir, path)):
raise Exception("path %s doesn't exist" % (path, ))
def check_not_exists(self, path):
if os.path.exists(os.path.join(os.environ["HOME"], ".aptly", path)):
if os.path.exists(os.path.join(os.environ["HOME"], self.aptlyDir, path)):
raise Exception("path %s exists" % (path, ))
def check_file_not_empty(self, path):
if os.stat(os.path.join(os.environ["HOME"], ".aptly", path))[6] == 0:
if os.stat(os.path.join(os.environ["HOME"], self.aptlyDir, path))[6] == 0:
raise Exception("file %s is empty" % (path, ))
def check_equal(self, a, b):
system/t12_api/publish.py
+88
View File
@@ -617,3 +617,91 @@ class PublishSwitchAPISkipCleanupTestRepo(APITest):
self.check_equal(self.delete_task("/api/publish/" + prefix + "/wheezy", params={"SkipCleanup": "1"}).json()['State'], TASK_SUCCEEDED)
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")
class ServePublishedTestRepo(APITest):
"""
GET /repos/:storage/*pkgPath
"""
def check(self):
d = self.random_name()
r = self.random_name()
f = "libboost-program-options-dev_1.62.0.1_i386.deb"
self.check_equal(self.upload("/api/files/" + d, f).status_code, 200)
self.check_equal(self.post("/api/repos", json={
"Name": r,
"Comment": "test repo",
"DefaultDistribution": r,
"DefaultComponent": "main"
}).status_code, 201)
self.check_equal(self.post(f"/api/repos/{r}/file/{d}").status_code, 200)
self.check_equal(self.post("/api/publish/filesystem:apiandserve:", json={
"SourceKind": "local",
"Sources": [
{
"Component": "main",
"Name": r
}
],
"Distribution": r,
"Signing": {
"Skip": True
}
}).status_code, 201)
get = self.get(f"/repos/apiandserve/pool/main/b/boost-defaults/{f}")
deb_content_types = [
"application/x-deb",
"application/x-debian-package",
"application/vnd.debian.binary-package"
]
if get.headers['content-type'] not in deb_content_types:
raise Exception(f"Received content-type {get.headers['content-type']} not one of expected: {deb_content_types}")
if len(get.content) != 3428:
raise Exception(f"Expected file size 3428 bytes != {get.status_code} bytes")
class ServePublishedNotFoundTestRepo(APITest):
"""
GET /repos/:storage/*pkgPath
"""
def check(self):
d = self.random_name()
r = self.random_name()
f = "libboost-program-options-dev_1.62.0.1_i386.deb"
self.check_equal(self.upload("/api/files/" + d, f).status_code, 200)
self.check_equal(self.post("/api/repos", json={
"Name": r,
"Comment": "test repo",
"DefaultDistribution": r,
"DefaultComponent": "main"
}).status_code, 201)
self.check_equal(self.post(f"/api/repos/{r}/file/{d}").status_code, 200)
self.check_equal(self.post("/api/publish/filesystem:apiandserve:", json={
"SourceKind": "local",
"Sources": [
{
"Component": "main",
"Name": r
}
],
"Distribution": r,
"Signing": {
"Skip": True
}
}).status_code, 201)
get = self.get("/repos/apiandserve/pool/main/b/boost-defaults/i-dont-exist")
if get.status_code != 404:
raise Exception(f"Expected status 404 != {get.status_code}")