almalinux/lorax
13
0
Fork 0
Code Issues Pull Requests Releases Activity

Add support for user defined package sources API

Browse Source 
This lives under /api/v0/projects/source/*

See the documentation for details

(cherry picked from commit 6d677b2207)
This commit is contained in:
Brian C. Lane 2018-05-29 12:22:34 -07:00
parent 5548f5f1c7
commit 8ac04a1521
13 changed files with 852 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

52
docs/lorax-composer.rst
View File

@ -235,3 +235,55 @@ the results directory, or it could do some post-processing on it. The end of
the function should always clean up the ``./compose/`` directory, removing any
unneeded extra files. This is especially true for the ``live-iso`` since it produces
the contents of the iso as well as the boot.iso itself.
Package Sources
---------------
By default lorax-composer uses the host's configured repositories. It copies
the ``*.repo`` files from ``/etc/yum.repos.d/`` into
``/var/lib/lorax/composer/repos.d/`` at startup, these are immutable system
repositories and cannot be deleted or changed. If you want to add additional
repos you can put them into ``/var/lib/lorax/composer/repos.d/`` or use the
``/api/v0/projects/source/*`` API routes to create them.
The new source can be added by doing a POST to the ``/api/v0/projects/source/new``
route using JSON (with `Content-Type` header set to `application/json`) or TOML
(with it set to `text/x-toml`). The format of the source looks like this (in
TOML)::
name = "custom-source-1"
url = "https://url/path/to/repository/"
type = "yum-baseurl"
proxy = "https://proxy-url/"
check_ssl = true
check_gpg = true
gpgkey_urls = ["https://url/path/to/gpg-key"]
The ``proxy`` and ``gpgkey_urls`` entries are optional. All of the others are required. The supported
types for the urls are:
* ``yum-baseurl`` is a URL to a yum repository.
* ``yum-mirrorlist`` is a URL for a mirrorlist.
* ``yum-metalink`` is a URL for a metalink.
If ``check_ssl`` is true the https certificates must be valid. If they are self-signed you can either set
this to false, or add your Certificate Authority to the host system.
If ``check_gpg`` is true the GPG key must either be installed on the host system, or ``gpgkey_urls``
should point to it.
You can edit an existing source (other than system sources), by doing a POST to the ``new`` route
with the new version of the source. It will overwrite the previous one.
A list of existing sources is available from ``/api/v0/projects/source/list``, and detailed info
on a source can be retrieved with the ``/api/v0/projects/source/info/<source-name>`` route. By default
it returns JSON but it can also return TOML if ``?format=toml`` is added to the request.
Non-system sources can be deleted by doing a ``DELETE`` request to the
``/api/v0/projects/source/delete/<source-name>`` route.
The documentation for the source API routes can be `found here <pylorax.api.html#api-v0-projects-source-list>`_
The configured sources are used for all blueprint depsolve operations, and for composing images.
When adding additional sources you must make sure that the packages in the source do not
conflict with any other package sources, otherwise depsolving will fail.

2
src/pylorax/api/config.py
View File

@ -44,9 +44,9 @@ def configure(conf_file="/etc/lorax/composer.conf", root_dir="/", test_config=Fa
conf.add_section("composer")
conf.set("composer", "share_dir", os.path.realpath(joinpaths(root_dir, "/usr/share/lorax/")))
conf.set("composer", "lib_dir", os.path.realpath(joinpaths(root_dir, "/var/lib/lorax/composer/")))
conf.set("composer", "repo_dir", os.path.realpath(joinpaths(root_dir, "/var/lib/lorax/composer/repos.d/")))
conf.set("composer", "dnf_conf", os.path.realpath(joinpaths(root_dir, "/var/tmp/composer/dnf.conf")))
conf.set("composer", "dnf_root", os.path.realpath(joinpaths(root_dir, "/var/tmp/composer/dnf/root/")))
conf.set("composer", "repo_dir", os.path.realpath(joinpaths(root_dir, "/var/tmp/composer/repos.d/")))
conf.set("composer", "cache_dir", os.path.realpath(joinpaths(root_dir, "/var/tmp/composer/cache/")))
conf.set("composer", "tmp", os.path.realpath(joinpaths(root_dir, "/var/tmp/")))

175
src/pylorax/api/projects.py
View File

@ -17,7 +17,10 @@
import logging
log = logging.getLogger("lorax-composer")
import os
from configparser import ConfigParser
import dnf
from glob import glob
import time
TIME_FORMAT = "%Y-%m-%dT%H:%M:%S"
@ -318,3 +321,175 @@ def modules_info(dbo, module_names):
module["dependencies"] = projects_depsolve(dbo, [(module["name"], "*.*")])
return modules
def repo_to_source(repo, system_source):
"""Return a Weldr Source dict created from the DNF Repository
:param repo: DNF Repository
:type repo: dnf.RepoDict
:param system_source: True if this source is an immutable system source
:type system_source: bool
:returns: A dict with Weldr Source fields filled in
:rtype: dict
Example::
{
"check_gpg": true,
"check_ssl": true,
"gpgkey_url": [
"file:///etc/pki/rpm-gpg/RPM-GPG-KEY-fedora-28-x86_64"
],
"name": "fedora",
"proxy": "http://proxy.brianlane.com:8123",
"system": true
"type": "yum-metalink",
"url": "https://mirrors.fedoraproject.org/metalink?repo=fedora-28&arch=x86_64"
}
"""
source = {"name": repo.id, "system": system_source}
if repo.baseurl:
source["url"] = repo.baseurl[0]
source["type"] = "yum-baseurl"
elif repo.metalink:
source["url"] = repo.metalink
source["type"] = "yum-metalink"
elif repo.mirrorlist:
source["url"] = repo.mirrorlist
source["type"] = "yum-mirrorlist"
else:
raise RuntimeError("Repo has no baseurl, metalink, or mirrorlist")
# proxy is optional
if repo.proxy:
source["proxy"] = repo.proxy
if not repo.sslverify:
source["check_ssl"] = False
else:
source["check_ssl"] = True
if not repo.gpgcheck:
source["check_gpg"] = False
else:
source["check_gpg"] = True
if repo.gpgkey:
source["gpgkey_urls"] = repo.gpgkey
return source
def source_to_repo(source, dnf_conf):
"""Return a dnf Repo object created from a source dict
:param source: A Weldr source dict
:type source: dict
:returns: A dnf Repo object
:rtype: dnf.Repo
Example::
{
"check_gpg": True,
"check_ssl": True,
"gpgkey_urls": [
"file:///etc/pki/rpm-gpg/RPM-GPG-KEY-fedora-28-x86_64"
],
"name": "fedora",
"proxy": "http://proxy.brianlane.com:8123",
"system": True
"type": "yum-metalink",
"url": "https://mirrors.fedoraproject.org/metalink?repo=fedora-28&arch=x86_64"
}
"""
repo = dnf.repo.Repo(source["name"], dnf_conf)
if source["type"] == "yum-baseurl":
repo.baseurl = [source["url"]]
elif source["type"] == "yum-metalink":
repo.metalink = source["url"]
elif source["type"] == "yum-mirrorlist":
repo.mirrorlist = source["url"]
if "proxy" in source:
repo.proxy = source["proxy"]
if source["check_ssl"]:
repo.sslverify = True
else:
repo.sslverify = False
if source["check_gpg"]:
repo.gpgcheck = True
else:
repo.gpgcheck = False
if "gpgkey_urls" in source:
repo.gpgkey = source["gpgkey_urls"]
repo.enable()
return repo
def get_source_ids(source_path):
"""Return a list of the source ids in a file
:param source_path: Full path and filename of the source (yum repo) file
:type source_path: str
:returns: A list of source id strings
:rtype: list of str
"""
if not os.path.exists(source_path):
return []
cfg = ConfigParser()
cfg.read(source_path)
return cfg.sections()
def get_repo_sources(source_glob):
"""Return a list of sources from a directory of yum repositories
:param source_glob: A glob to use to match the source files, including full path
:type source_glob: str
:returns: A list of the source ids in all of the matching files
:rtype: list of str
"""
sources = []
for f in glob(source_glob):
sources.extend(get_source_ids(f))
return sources
def delete_repo_source(source_glob, source_name):
"""Delete a source from a repo file
:param source_glob: A glob of the repo sources to search
:type source_glob: str
:returns: None
:raises: ProjectsError if there was a problem
A repo file may have multiple sources in it, delete only the selected source.
If it is the last one in the file, delete the file.
WARNING: This will delete ANY source, the caller needs to ensure that a system
source_name isn't passed to it.
"""
found = False
for f in glob(source_glob):
try:
cfg = ConfigParser()
cfg.read(f)
if source_name in cfg.sections():
found = True
cfg.remove_section(source_name)
# If there are other sections, rewrite the file without the deleted one
if len(cfg.sections()) > 0:
with open(f, "w") as cfg_file:
cfg.write(cfg_file)
else:
# No sections left, just delete the file
os.unlink(f)
except Exception as e:
raise ProjectsError("Problem deleting repo source %s: %s" % (source_name, str(e)))
if not found:
raise ProjectsError("source %s not found" % source_name)

4
src/pylorax/api/queue.py
View File

@ -90,7 +90,7 @@ def monitor(cfg):
# Pick the oldest and move it into ./run/
if not uuids:
# No composes left to process, sleep for a bit
time.sleep(30)
time.sleep(5)
else:
src = joinpaths(cfg.composer_dir, "queue/new", uuids[0])
dst = joinpaths(cfg.composer_dir, "queue/run", uuids[0])
@ -196,7 +196,7 @@ def make_compose(cfg, results_dir):
test_path = joinpaths(results_dir, "TEST")
if os.path.exists(test_path):
# Pretend to run the compose
time.sleep(10)
time.sleep(5)
try:
test_mode = int(open(test_path, "r").read())
except Exception:

208
src/pylorax/api/v0.py
View File

@ -501,6 +501,94 @@ POST `/api/v0/blueprints/tag/<blueprint_name>`
]
}
`/api/v0/projects/source/list`
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
Return the list of repositories used for depsolving and installing packages.
Example::
{
"sources": [
"fedora",
"fedora-cisco-openh264",
"fedora-updates-testing",
"fedora-updates"
]
}
`/api/v0/projects/source/info/<source-names>`
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
Return information about the comma-separated list of source names. Or all of the
sources if '*' is passed. Note that general globbing is not supported, only '*'.
immutable system sources will have the "system" field set to true. User added sources
will have it set to false. System sources cannot be changed or deleted.
Example::
{
"errors": [],
"sources": {
"fedora": {
"check_gpg": true,
"check_ssl": true,
"gpgkey_urls": [
"file:///etc/pki/rpm-gpg/RPM-GPG-KEY-fedora-28-x86_64"
],
"name": "fedora",
"proxy": "http://proxy.brianlane.com:8123",
"system": true,
"type": "yum-metalink",
"url": "https://mirrors.fedoraproject.org/metalink?repo=fedora-28&arch=x86_64"
}
}
}
POST `/api/v0/projects/source/new`
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
Add (or change) a source for use when depsolving blueprints and composing images.
The ``proxy`` and ``gpgkey_urls`` entries are optional. All of the others are required. The supported
types for the urls are:
* ``yum-baseurl`` is a URL to a yum repository.
* ``yum-mirrorlist`` is a URL for a mirrorlist.
* ``yum-metalink`` is a URL for a metalink.
If ``check_ssl`` is true the https certificates must be valid. If they are self-signed you can either set
this to false, or add your Certificate Authority to the host system.
If ``check_gpg`` is true the GPG key must either be installed on the host system, or ``gpgkey_urls``
should point to it.
You can edit an existing source (other than system sources), by doing a POST
of the new version of the source. It will overwrite the previous one.
Example::
{
"name": "custom-source-1",
"url": "https://url/path/to/repository/",
"type": "yum-baseurl",
"check_ssl": true,
"check_gpg": true,
"gpgkey_urls": [
"https://url/path/to/gpg-key"
]
}
DELETE `/api/v0/projects/source/delete/<source-name>`
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
Delete a user added source. This will fail if a system source is passed to
it.
The response will be a status response with `status` set to true, or an
error response with it set to false and an error message included.
`/api/v0/modules/list[?offset=0&limit=20]`
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
@ -878,11 +966,14 @@ log = logging.getLogger("lorax-composer")
import os
from flask import jsonify, request, Response, send_file
import pytoml as toml
from pylorax.sysutils import joinpaths
from pylorax.api.compose import start_build, compose_types
from pylorax.api.crossdomain import crossdomain
from pylorax.api.projects import projects_list, projects_info, projects_depsolve
from pylorax.api.projects import modules_list, modules_info, ProjectsError
from pylorax.api.projects import modules_list, modules_info, ProjectsError, repo_to_source
from pylorax.api.projects import get_repo_sources, delete_repo_source, source_to_repo
from pylorax.api.queue import queue_status, build_status, uuid_delete, uuid_status, uuid_info
from pylorax.api.queue import uuid_tar, uuid_image, uuid_cancel, uuid_log
from pylorax.api.recipes import list_branch_files, read_recipe_commit, recipe_filename, list_commits
@ -1301,6 +1392,121 @@ def v0_api(api):
return jsonify(projects=deps)
@api.route("/api/v0/projects/source/list")
@crossdomain(origin="*")
def v0_projects_source_list():
"""Return the list of source names"""
with api.config["DNFLOCK"].lock:
repos = list(api.config["DNFLOCK"].dbo.repos.iter_enabled())
sources = sorted([r.id for r in repos])
return jsonify(sources=sources)
@api.route("/api/v0/projects/source/info/<source_names>")
@crossdomain(origin="*")
def v0_projects_source_info(source_names):
"""Return detailed info about the list of sources"""
out_fmt = request.args.get("format", "json")
# Return info on all of the sources
if source_names == "*":
with api.config["DNFLOCK"].lock:
source_names = ",".join(r.id for r in api.config["DNFLOCK"].dbo.repos.iter_enabled())
sources = {}
errors = []
system_sources = get_repo_sources("/etc/yum.repos.d/*.repo")
for source in source_names.split(","):
with api.config["DNFLOCK"].lock:
repo = api.config["DNFLOCK"].dbo.repos.get(source, None)
if not repo:
errors.append("%s is not a valid source" % source)
continue
sources[repo.id] = repo_to_source(repo, repo.id in system_sources)
if out_fmt == "toml":
# With TOML output we just want to dump the raw sources, skipping the errors
return toml.dumps(sources)
else:
return jsonify(sources=sources, errors=errors)
@api.route("/api/v0/projects/source/new", methods=["POST"])
@crossdomain(origin="*")
def v0_projects_source_new():
"""Add a new package source. Or change an existing one"""
if request.headers['Content-Type'] == "text/x-toml":
source = toml.loads(request.data)
else:
source = request.get_json(cache=False)
system_sources = get_repo_sources("/etc/yum.repos.d/*.repo")
if source["name"] in system_sources:
return jsonify(status=False, errors=["%s is a system source, it cannot be deleted." % source["name"]]), 400
try:
# Remove it from the RepoDict (NOTE that this isn't explicitly supported by the DNF API)
with api.config["DNFLOCK"].lock:
dbo = api.config["DNFLOCK"].dbo
# If this repo already exists, delete it and replace it with the new one
repos = list(r.id for r in dbo.repos.iter_enabled())
if source["name"] in repos:
del dbo.repos[source["name"]]
# XXX - BCL DIAGNOSTIC
repos = list(r.id for r in dbo.repos.iter_enabled())
if source["name"] in repos:
return jsonify(status=False, errors=["Failed to delete DNF repo %s" % source["name"]]), 400
repo = source_to_repo(source, dbo.conf)
dbo.repos.add(repo)
log.info("Updating repository metadata after adding %s", source["name"])
dbo.fill_sack(load_system_repo=False)
dbo.read_comps()
# Write the new repo to disk, replacing any existing ones
repo_dir = api.config["COMPOSER_CFG"].get("composer", "repo_dir")
# Remove any previous sources with this name, ignore it if it isn't found
try:
delete_repo_source(joinpaths(repo_dir, "*.repo"), source["name"])
except ProjectsError:
pass
# Make sure the source name can't contain a path traversal by taking the basename
source_path = joinpaths(repo_dir, os.path.basename("%s.repo" % source["name"]))
with open(source_path, "w") as f:
f.write(str(repo))
except Exception as e:
return jsonify(status=False, errors=[str(e)]), 400
return jsonify(status=True)
@api.route("/api/v0/projects/source/delete/<source_name>", methods=["DELETE"])
@crossdomain(origin="*")
def v0_projects_source_delete(source_name):
"""Delete the named source and return a status response"""
system_sources = get_repo_sources("/etc/yum.repos.d/*.repo")
if source_name in system_sources:
return jsonify(status=False, errors=["%s is a system source, it cannot be deleted." % source_name]), 400
share_dir = api.config["COMPOSER_CFG"].get("composer", "repo_dir")
try:
# Remove the file entry for the source
delete_repo_source(joinpaths(share_dir, "*.repo"), source_name)
# Remove it from the RepoDict (NOTE that this isn't explicitly supported by the DNF API)
with api.config["DNFLOCK"].lock:
if source_name in api.config["DNFLOCK"].dbo.repos:
del api.config["DNFLOCK"].dbo.repos[source_name]
log.info("Updating repository metadata after removing %s", source_name)
api.config["DNFLOCK"].dbo.fill_sack(load_system_repo=False)
api.config["DNFLOCK"].dbo.read_comps()
except ProjectsError as e:
log.error("(v0_projects_source_delete) %s", str(e))
return jsonify(status=False, errors=[str(e)]), 400
return jsonify(status=True)
@api.route("/api/v0/modules/list")
@api.route("/api/v0/modules/list/<module_names>")
@crossdomain(origin="*")

47
tests/pylorax/repos/multiple.repo Normal file
View File

@ -0,0 +1,47 @@
[lorax-1]
name=Lorax test repo 1
failovermethod=priority
baseurl=file:///tmp/lorax-empty-repo/
enabled=1
metadata_expire=7d
repo_gpgcheck=0
type=rpm
gpgcheck=1
gpgkey=file:///etc/pki/rpm-gpg/RPM-GPG-KEY-fedora-$releasever-$basearch
skip_if_unavailable=False
[lorax-2]
name=Lorax test repo 2
failovermethod=priority
baseurl=file:///tmp/lorax-empty-repo/
enabled=1
metadata_expire=7d
repo_gpgcheck=0
type=rpm
gpgcheck=1
gpgkey=file:///etc/pki/rpm-gpg/RPM-GPG-KEY-fedora-$releasever-$basearch
skip_if_unavailable=False
[lorax-3]
name=Lorax test repo 3
failovermethod=priority
baseurl=file:///tmp/lorax-empty-repo/
enabled=1
metadata_expire=7d
repo_gpgcheck=0
type=rpm
gpgcheck=1
gpgkey=file:///etc/pki/rpm-gpg/RPM-GPG-KEY-fedora-$releasever-$basearch
skip_if_unavailable=False
[lorax-4]
name=Lorax test repo 4
failovermethod=priority
baseurl=file:///tmp/lorax-empty-repo/
enabled=1
metadata_expire=7d
repo_gpgcheck=0
type=rpm
gpgcheck=1
gpgkey=file:///etc/pki/rpm-gpg/RPM-GPG-KEY-fedora-$releasever-$basearch
skip_if_unavailable=False

11
tests/pylorax/repos/other.repo Normal file
View File

@ -0,0 +1,11 @@
[other-repo]
name=Other repo
failovermethod=priority
baseurl=file:///tmp/lorax-empty-repo/
enabled=1
metadata_expire=7d
repo_gpgcheck=0
type=rpm
gpgcheck=1
gpgkey=file:///etc/pki/rpm-gpg/RPM-GPG-KEY-fedora-$releasever-$basearch
skip_if_unavailable=False

11
tests/pylorax/repos/single.repo Normal file
View File

@ -0,0 +1,11 @@
[single-repo]
name=One repo in the file
failovermethod=priority
baseurl=file:///tmp/lorax-empty-repo/
enabled=1
metadata_expire=7d
repo_gpgcheck=0
type=rpm
gpgcheck=1
gpgkey=file:///etc/pki/rpm-gpg/RPM-GPG-KEY-fedora-$releasever-$basearch
skip_if_unavailable=False

6
tests/pylorax/source/replace-repo.toml Normal file
View File

@ -0,0 +1,6 @@
name = "single-repo"
url = "file:///tmp/lorax-empty-repo/"
type = "yum-baseurl"
check_ssl = false
check_gpg = true
gpgkey_urls = []

1
tests/pylorax/source/test-repo.json Normal file
View File

@ -0,0 +1 @@
{"name": "new-repo-1", "url": "file:///tmp/lorax-empty-repo/", "type": "yum-baseurl", "check_ssl": true, "check_gpg": true, "gpgkey_urls": ["file:///etc/pki/rpm-gpg/RPM-GPG-KEY-fedora-$releasever-$basearch"]}

6
tests/pylorax/source/test-repo.toml Normal file
View File

@ -0,0 +1,6 @@
name = "new-repo-2"
url = "file:///tmp/lorax-empty-repo/"
type = "yum-baseurl"
check_ssl = true
check_gpg = true
gpgkey_urls = ["file:///etc/pki/rpm-gpg/RPM-GPG-KEY-fedora-$releasever-$basearch"]

218
tests/pylorax/test_projects.py
View File

@ -14,16 +14,20 @@
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
import dnf
from glob import glob
import os
import shutil
import tempfile
import time
import unittest
from pylorax.sysutils import joinpaths
from pylorax.api.config import configure, make_dnf_dirs
from pylorax.api.projects import api_time, api_changelog, pkg_to_project, pkg_to_project_info, pkg_to_dep
from pylorax.api.projects import proj_to_module, projects_list, projects_info, projects_depsolve
from pylorax.api.projects import modules_list, modules_info, ProjectsError, dep_evra, dep_nevra
from pylorax.api.projects import repo_to_source, get_repo_sources, delete_repo_source, source_to_repo
from pylorax.api.dnfbase import get_base_object
class Package(object):
@ -209,3 +213,217 @@ class ConfigureTest(unittest.TestCase):
def test_configure_reads_non_existing_file(self):
config = configure(conf_file=self.conf_file + '.non-existing')
self.assertEqual(config.get('composer', 'cache_dir'), '/var/tmp/composer/cache')
class FakeRepoBaseUrl():
id = "fake-repo-baseurl"
baseurl = ["https://fake-repo.base.url"]
metalink = ""
mirrorlist = ""
proxy = ""
sslverify = True
gpgcheck = True
gpgkey = []
def fakerepo_baseurl():
return {
"check_gpg": True,
"check_ssl": True,
"name": "fake-repo-baseurl",
"system": False,
"type": "yum-baseurl",
"url": "https://fake-repo.base.url"
}
class FakeSystemRepo():
id = "fake-system-repo"
baseurl = ["https://fake-repo.base.url"]
metalink = ""
mirrorlist = ""
proxy = ""
sslverify = True
gpgcheck = True
gpgkey = []
def fakesystem_repo():
return {
"check_gpg": True,
"check_ssl": True,
"name": "fake-system-repo",
"system": True,
"type": "yum-baseurl",
"url": "https://fake-repo.base.url"
}
class FakeRepoMetalink():
id = "fake-repo-metalink"
baseurl = []
metalink = "https://fake-repo.metalink"
proxy = ""
sslverify = True
gpgcheck = True
gpgkey = []
def fakerepo_metalink():
return {
"check_gpg": True,
"check_ssl": True,
"name": "fake-repo-metalink",
"system": False,
"type": "yum-metalink",
"url": "https://fake-repo.metalink"
}
class FakeRepoMirrorlist():
id = "fake-repo-mirrorlist"
baseurl = []
metalink = ""
mirrorlist = "https://fake-repo.mirrorlist"
proxy = ""
sslverify = True
gpgcheck = True
gpgkey = []
def fakerepo_mirrorlist():
return {
"check_gpg": True,
"check_ssl": True,
"name": "fake-repo-mirrorlist",
"system": False,
"type": "yum-mirrorlist",
"url": "https://fake-repo.mirrorlist"
}
class FakeRepoProxy():
id = "fake-repo-proxy"
baseurl = ["https://fake-repo.base.url"]
metalink = ""
mirrorlist = ""
proxy = "https://fake-repo.proxy"
sslverify = True
gpgcheck = True
gpgkey = []
def fakerepo_proxy():
return {
"check_gpg": True,
"check_ssl": True,
"name": "fake-repo-proxy",
"proxy": "https://fake-repo.proxy",
"system": False,
"type": "yum-baseurl",
"url": "https://fake-repo.base.url"
}
class FakeRepoGPGKey():
id = "fake-repo-gpgkey"
baseurl = ["https://fake-repo.base.url"]
metalink = ""
mirrorlist = ""
proxy = ""
sslverify = True
gpgcheck = True
gpgkey = ["https://fake-repo.gpgkey"]
def fakerepo_gpgkey():
return {
"check_gpg": True,
"check_ssl": True,
"gpgkey_urls": [
"https://fake-repo.gpgkey"
],
"name": "fake-repo-gpgkey",
"system": False,
"type": "yum-baseurl",
"url": "https://fake-repo.base.url"
}
class SourceTest(unittest.TestCase):
@classmethod
def setUpClass(self):
self.tmp_dir = tempfile.mkdtemp(prefix="lorax.test.repo.")
for f in glob("./tests/pylorax/repos/*.repo"):
shutil.copy2(f, self.tmp_dir)
self.dbo = dnf.Base()
@classmethod
def tearDownClass(self):
shutil.rmtree(self.tmp_dir)
def test_repo_to_source_baseurl(self):
"""Test a repo with a baseurl"""
self.assertEqual(repo_to_source(FakeRepoBaseUrl(), False), fakerepo_baseurl())
def test_system_repo(self):
"""Test a system repo with a baseurl"""
self.assertEqual(repo_to_source(FakeSystemRepo(), True), fakesystem_repo())
def test_repo_to_source_metalink(self):
"""Test a repo with a metalink"""
self.assertEqual(repo_to_source(FakeRepoMetalink, False), fakerepo_metalink())
def test_repo_to_source_mirrorlist(self):
"""Test a repo with a mirrorlist"""
self.assertEqual(repo_to_source(FakeRepoMirrorlist, False), fakerepo_mirrorlist())
def test_repo_to_source_proxy(self):
"""Test a repo with a proxy"""
self.assertEqual(repo_to_source(FakeRepoProxy, False), fakerepo_proxy())
def test_repo_to_source_gpgkey(self):
"""Test a repo with a GPG key"""
self.assertEqual(repo_to_source(FakeRepoGPGKey, False), fakerepo_gpgkey())
def test_get_repo_sources(self):
"""Test getting a list of sources from a repo directory"""
sources = get_repo_sources(joinpaths(self.tmp_dir, "*.repo"))
self.assertTrue("lorax-1" in sources)
self.assertTrue("lorax-2" in sources)
def test_delete_source_multiple(self):
"""Test deleting a source from a repo file with multiple entries"""
delete_repo_source(joinpaths(self.tmp_dir, "*.repo"), "lorax-3")
sources = get_repo_sources(joinpaths(self.tmp_dir, "*.repo"))
self.assertTrue("lorax-3" not in sources)
def test_delete_source_single(self):
"""Test deleting a source from a repo with only 1 entry"""
delete_repo_source(joinpaths(self.tmp_dir, "*.repo"), "single-repo")
sources = get_repo_sources(joinpaths(self.tmp_dir, "*.repo"))
self.assertTrue("single-repo" not in sources)
self.assertTrue(not os.path.exists(joinpaths(self.tmp_dir, "single.repo")))
def test_delete_source_other(self):
"""Test deleting a source from a repo that doesn't match the source name"""
with self.assertRaises(ProjectsError):
delete_repo_source(joinpaths(self.tmp_dir, "*.repo"), "unknown-source")
sources = get_repo_sources(joinpaths(self.tmp_dir, "*.repo"))
self.assertTrue("lorax-1" in sources)
self.assertTrue("lorax-2" in sources)
self.assertTrue("lorax-4" in sources)
self.assertTrue("other-repo" in sources)
def test_source_to_repo_baseurl(self):
"""Test creating a dnf.Repo with a baseurl"""
repo = source_to_repo(fakerepo_baseurl(), self.dbo.conf)
self.assertEqual(repo.baseurl[0], fakerepo_baseurl()["url"])
def test_source_to_repo_metalink(self):
"""Test creating a dnf.Repo with a metalink"""
repo = source_to_repo(fakerepo_metalink(), self.dbo.conf)
self.assertEqual(repo.metalink, fakerepo_metalink()["url"])
def test_source_to_repo_mirrorlist(self):
"""Test creating a dnf.Repo with a mirrorlist"""
repo = source_to_repo(fakerepo_mirrorlist(), self.dbo.conf)
self.assertEqual(repo.mirrorlist, fakerepo_mirrorlist()["url"])
def test_source_to_repo_proxy(self):
"""Test creating a dnf.Repo with a proxy"""
repo = source_to_repo(fakerepo_proxy(), self.dbo.conf)
self.assertEqual(repo.proxy, fakerepo_proxy()["proxy"])
def test_source_to_repo_gpgkey(self):
"""Test creating a dnf.Repo with a proxy"""
repo = source_to_repo(fakerepo_gpgkey(), self.dbo.conf)
self.assertEqual(repo.gpgkey, fakerepo_gpgkey()["gpgkey_urls"])

116
tests/pylorax/test_server.py
View File

@ -49,6 +49,18 @@ class ServerTestCase(unittest.TestCase):
raise RuntimeError("\n".join(errors))
make_dnf_dirs(server.config["COMPOSER_CFG"])
# copy over the test dnf repositories
dnf_repo_dir = server.config["COMPOSER_CFG"].get("composer", "repo_dir")
os.makedirs(dnf_repo_dir)
for f in glob("./tests/pylorax/repos/*.repo"):
shutil.copy2(f, dnf_repo_dir)
# dnf repo baseurl has to point to an absolute directory, so we use /tmp/lorax-empty-repo/ in the files
# and create an empty repository
os.makedirs("/tmp/lorax-empty-repo/")
os.system("createrepo_c /tmp/lorax-empty-repo/")
dbo = get_base_object(server.config["COMPOSER_CFG"])
server.config["DNFLOCK"] = DNFLock(dbo=dbo, lock=Lock())
@ -71,6 +83,7 @@ class ServerTestCase(unittest.TestCase):
@classmethod
def tearDownClass(self):
shutil.rmtree(server.config["REPO_DIR"])
shutil.rmtree("/tmp/lorax-empty-repo/")
def test_01_status(self):
"""Test the /api/status route"""
@ -469,6 +482,107 @@ class ServerTestCase(unittest.TestCase):
self.assertEqual(len(deps) > 10, True)
self.assertEqual(deps[0]["name"], "basesystem")
def test_projects_source_00_list(self):
"""Test /api/v0/projects/source/list"""
resp = self.server.get("/api/v0/projects/source/list")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
self.assertEqual(data["sources"], ["fedora", "lorax-1", "lorax-2", "lorax-3", "lorax-4", "other-repo", "single-repo", "updates"])
def test_projects_source_00_info(self):
"""Test /api/v0/projects/source/info"""
resp = self.server.get("/api/v0/projects/source/info/single-repo")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
sources = data["sources"]
self.assertTrue("single-repo" in sources)
def test_projects_source_00_new_json(self):
"""Test /api/v0/projects/source/new with a new json source"""
json_source = open("./tests/pylorax/source/test-repo.json").read()
self.assertTrue(len(json_source) > 0)
resp = self.server.post("/api/v0/projects/source/new",
data=json_source,
content_type="application/json")
data = json.loads(resp.data)
self.assertEqual(data, {"status":True})
def test_projects_source_00_new_toml(self):
"""Test /api/v0/projects/source/new with a new toml source"""
toml_source = open("./tests/pylorax/source/test-repo.toml").read()
self.assertTrue(len(toml_source) > 0)
resp = self.server.post("/api/v0/projects/source/new",
data=toml_source,
content_type="text/x-toml")
data = json.loads(resp.data)
self.assertEqual(data, {"status":True})
def test_projects_source_00_replace(self):
"""Test /api/v0/projects/source/new with a replacement source"""
toml_source = open("./tests/pylorax/source/replace-repo.toml").read()
self.assertTrue(len(toml_source) > 0)
resp = self.server.post("/api/v0/projects/source/new",
data=toml_source,
content_type="text/x-toml")
data = json.loads(resp.data)
self.assertEqual(data, {"status":True})
# Check to see if it was really changed
resp = self.server.get("/api/v0/projects/source/info/single-repo")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
sources = data["sources"]
self.assertTrue("single-repo" in sources)
repo = sources["single-repo"]
self.assertEqual(repo["check_ssl"], False)
self.assertTrue("gpgkey_urls" not in repo)
def test_projects_source_01_delete_system(self):
"""Test /api/v0/projects/source/delete a system source"""
resp = self.server.delete("/api/v0/projects/source/delete/fedora")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
self.assertEqual(data["status"], False)
# Make sure fedora is still listed
resp = self.server.get("/api/v0/projects/source/list")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
self.assertTrue("fedora" in data["sources"])
def test_projects_source_02_delete_single(self):
"""Test /api/v0/projects/source/delete a single source"""
resp = self.server.delete("/api/v0/projects/source/delete/single-repo")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
self.assertEqual(data, {"status":True})
# Make sure single-repo isn't listed
resp = self.server.get("/api/v0/projects/source/list")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
self.assertTrue("single-repo" not in data["sources"])
def test_projects_source_03_delete_unknown(self):
"""Test /api/v0/projects/source/delete an unknown source"""
resp = self.server.delete("/api/v0/projects/source/delete/unknown-repo")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
self.assertEqual(data["status"], False)
def test_projects_source_04_delete_multi(self):
"""Test /api/v0/projects/source/delete a source from a file with multiple sources"""
resp = self.server.delete("/api/v0/projects/source/delete/lorax-3")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
self.assertEqual(data, {"status":True})
# Make sure single-repo isn't listed
resp = self.server.get("/api/v0/projects/source/list")
data = json.loads(resp.data)
self.assertNotEqual(data, None)
self.assertTrue("lorax-3" not in data["sources"])
def test_modules_list(self):
"""Test /api/v0/modules/list"""
resp = self.server.get("/api/v0/modules/list")
@ -561,7 +675,7 @@ class ServerTestCase(unittest.TestCase):
return True
if time.time() > start + 60:
return False
time.sleep(5)
time.sleep(1)
def test_compose_01_types(self):
"""Test the /api/v0/compose/types route"""