sipwise
/
repoapi
mirror of https://github.com/sipwise/repoapi.git
1
0
Fork 0
Code Issues Releases Wiki Activity

TT#121955 migrate to pathlib

Browse Source 
Change-Id: Ie3dbd8503d6c61dda3d187de65a5694cbc4f957f
pull/5/head
Victor Seva 4 years ago
parent df32d4377d
commit d832e3233c
13 changed files with 100 additions and 117 deletions
Show Stats Download Patch File Download Diff File

16
build/utils.py
Unescape View File

@ -1,4 +1,4 @@
# Copyright (C) 2017-2020 The Sipwise Team - http://sipwise.com # Copyright (C) 2017-2022 The Sipwise Team - http://sipwise.com
# #
# This program is free software: you can redistribute it and/or modify it # This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU General Public License as published by the Free # under the terms of the GNU General Public License as published by the Free
@ -13,9 +13,9 @@
# You should have received a copy of the GNU General Public License along # You should have received a copy of the GNU General Public License along
# with this program. If not, see <http://www.gnu.org/licenses/>. # with this program. If not, see <http://www.gnu.org/licenses/>.
import logging import logging
import os
import re import re
import urllib import urllib
from os import walk
from pathlib import Path from pathlib import Path
from uuid import uuid4 from uuid import uuid4
@ -180,15 +180,11 @@ class ReleaseConfig(object):
def supported_releases(cls): def supported_releases(cls):
skip_files = ["{}.yml".format(x) for x in settings.BUILD_RELEASES_SKIP] skip_files = ["{}.yml".format(x) for x in settings.BUILD_RELEASES_SKIP]
res = [] res = []
for root, dirs, files in os.walk( for root, dirs, files in walk(settings.BUILD_REPOS_SCRIPTS_CONFIG_DIR):
settings.BUILD_REPOS_SCRIPTS_CONFIG_DIR
):
if "trunk.yml" in files: if "trunk.yml" in files:
files.remove("trunk.yml") files.remove("trunk.yml")
cfg = cls.load_config( cfg = cls.load_config(
os.path.join( settings.BUILD_REPOS_SCRIPTS_CONFIG_DIR / "trunk.yml"
settings.BUILD_REPOS_SCRIPTS_CONFIG_DIR, "trunk.yml"
)
) )
for dist in cfg["distris"]: for dist in cfg["distris"]:
res.append(dist) res.append(dist)
@ -217,8 +213,8 @@ class ReleaseConfig(object):
if filename is None: if filename is None:
filename = name filename = name
self.config_file = "{}.yml".format(filename) self.config_file = "{}.yml".format(filename)
self.config_path = os.path.join( self.config_path = (
settings.BUILD_REPOS_SCRIPTS_CONFIG_DIR, self.config_file settings.BUILD_REPOS_SCRIPTS_CONFIG_DIR / self.config_file
) )
self.config = self.load_config(self.config_path) self.config = self.load_config(self.config_path)
try: try:

23
release_changed/test_tasks.py
Unescape View File

@ -1,4 +1,4 @@
# Copyright (C) 2020 The Sipwise Team - http://sipwise.com # Copyright (C) 2020-2022 The Sipwise Team - http://sipwise.com
# #
# This program is free software: you can redistribute it and/or modify it # This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU General Public License as published by the Free # under the terms of the GNU General Public License as published by the Free
@ -12,7 +12,6 @@
# #
# You should have received a copy of the GNU General Public License along # You should have received a copy of the GNU General Public License along
# with this program. If not, see <http://www.gnu.org/licenses/>. # with this program. If not, see <http://www.gnu.org/licenses/>.
from os.path import join
from unittest.mock import mock_open from unittest.mock import mock_open
from unittest.mock import patch from unittest.mock import patch
@ -22,13 +21,11 @@ from .models import ReleaseChanged
from repoapi.models import JenkinsBuildInfo from repoapi.models import JenkinsBuildInfo
from repoapi.test.base import BaseTest from repoapi.test.base import BaseTest
FIXTURES_PATH = join(settings.BASE_DIR, "release_changed", "fixtures") FIXTURES_PATH = settings.BASE_DIR.joinpath("release_changed", "fixtures")
FILE_PATH = join(FIXTURES_PATH, "test_envVars.json") FILE_PATH = FIXTURES_PATH / "test_envVars.json"
FILE_PATH_DONE = join(FIXTURES_PATH, "test_envVars_done.json") FILE_PATH_DONE = FIXTURES_PATH / "test_envVars_done.json"
with open(FILE_PATH) as file: DATA = FILE_PATH.read_text()
DATA = file.read() DATA_DONE = FILE_PATH_DONE.read_text()
with open(FILE_PATH_DONE) as file:
DATA_DONE = file.read()
class TasksTestCase(BaseTest): class TasksTestCase(BaseTest):
@ -37,7 +34,7 @@ class TasksTestCase(BaseTest):
def test_process_create(self): def test_process_create(self):
jbi = JenkinsBuildInfo.objects.get(pk=1) jbi = JenkinsBuildInfo.objects.get(pk=1)
with patch("builtins.open", mock_open(read_data=DATA)): with patch("builtins.open", mock_open(read_data=DATA)):
tasks.process_result.delay(jbi.id, FILE_PATH) tasks.process_result.delay(jbi.id, str(FILE_PATH))
r = ReleaseChanged.objects.get( r = ReleaseChanged.objects.get(
label="base", version="mr8.5.1", vmtype="CE" label="base", version="mr8.5.1", vmtype="CE"
) )
@ -50,7 +47,7 @@ class TasksTestCase(BaseTest):
r_id = r.id r_id = r.id
jbi = JenkinsBuildInfo.objects.get(pk=1) jbi = JenkinsBuildInfo.objects.get(pk=1)
with patch("builtins.open", mock_open(read_data=DATA)): with patch("builtins.open", mock_open(read_data=DATA)):
tasks.process_result.delay(jbi.id, FILE_PATH) tasks.process_result.delay(jbi.id, str(FILE_PATH))
r = ReleaseChanged.objects.get( r = ReleaseChanged.objects.get(
label="base", version="mr8.5.1", vmtype="CE" label="base", version="mr8.5.1", vmtype="CE"
) )
@ -63,7 +60,7 @@ class TasksTestCase(BaseTest):
) )
jbi = JenkinsBuildInfo.objects.get(pk=1) jbi = JenkinsBuildInfo.objects.get(pk=1)
with patch("builtins.open", mock_open(read_data=DATA_DONE)): with patch("builtins.open", mock_open(read_data=DATA_DONE)):
tasks.process_result.delay(jbi.id, FILE_PATH) tasks.process_result.delay(jbi.id, str(FILE_PATH))
rs = ReleaseChanged.objects.filter( rs = ReleaseChanged.objects.filter(
label="base", version="mr8.5.1", vmtype="CE" label="base", version="mr8.5.1", vmtype="CE"
) )
@ -72,7 +69,7 @@ class TasksTestCase(BaseTest):
def test_process_done_no_obj(self): def test_process_done_no_obj(self):
jbi = JenkinsBuildInfo.objects.get(pk=1) jbi = JenkinsBuildInfo.objects.get(pk=1)
with patch("builtins.open", mock_open(read_data=DATA_DONE)): with patch("builtins.open", mock_open(read_data=DATA_DONE)):
tasks.process_result.delay(jbi.id, FILE_PATH) tasks.process_result.delay(jbi.id, str(FILE_PATH))
rs = ReleaseChanged.objects.filter( rs = ReleaseChanged.objects.filter(
label="base", version="mr8.5.1", vmtype="CE" label="base", version="mr8.5.1", vmtype="CE"
) )

11
repoapi/models/jbi.py
Unescape View File

@ -18,7 +18,6 @@ import re
from collections import OrderedDict from collections import OrderedDict
from datetime import datetime from datetime import datetime
from datetime import timedelta from datetime import timedelta
from os.path import join
from urllib.parse import urlparse from urllib.parse import urlparse
import structlog import structlog
@ -252,11 +251,13 @@ class JenkinsBuildInfo(models.Model):
@property @property
def build_path(self): def build_path(self):
return join(settings.JBI_BASEDIR, self.jobname, str(self.buildnumber)) return settings.JBI_BASEDIR.joinpath(
self.jobname, str(self.buildnumber)
)
@property @property
def build_info(self): def build_info(self):
path = join(self.build_path, "build.json") path = self.build_path.joinpath("build.json")
try: try:
with open(path, "r") as data_file: with open(path, "r") as data_file:
data = json.load(data_file) data = json.load(data_file)
@ -278,8 +279,8 @@ class JenkinsBuildInfo(models.Model):
return getattr(self, "_source") return getattr(self, "_source")
except AttributeError: except AttributeError:
pass pass
path = join(self.build_path, "artifact") path = self.build_path.joinpath("artifact")
dscs = glob.glob(join(path, "*.dsc")) dscs = glob.glob(str(path.joinpath("*.dsc")))
if len(dscs) != 1: if len(dscs) != 1:
logger.error("more than one dsc file on artifact dir", path=path) logger.error("more than one dsc file on artifact dir", path=path)
return None return None

12
repoapi/settings/common.py
Unescape View File

@ -1,4 +1,4 @@
# Copyright (C) 2015-2020 The Sipwise Team - http://sipwise.com # Copyright (C) 2015-2022 The Sipwise Team - http://sipwise.com
# #
# This program is free software: you can redistribute it and/or modify it # This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU General Public License as published by the Free # under the terms of the GNU General Public License as published by the Free
@ -13,13 +13,13 @@
# You should have received a copy of the GNU General Public License along # You should have received a copy of the GNU General Public License along
# with this program. If not, see <http://www.gnu.org/licenses/>. # with this program. If not, see <http://www.gnu.org/licenses/>.
# #
# Build paths inside the project like this: os.path.join(BASE_DIR, ...)
import os import os
from os.path import dirname from pathlib import Path
import structlog import structlog
BASE_DIR = dirname(dirname(dirname(os.path.abspath(__file__)))) # Build paths inside the project like this: BASE_DIR / 'subdir'.
BASE_DIR = Path(__file__).resolve(strict=True).parent.parent.parent
INSTALLED_APPS = [ INSTALLED_APPS = [
"django.contrib.admin", "django.contrib.admin",
@ -105,7 +105,7 @@ STATICFILES_FINDERS = (
"django_assets.finders.AssetsFinder", "django_assets.finders.AssetsFinder",
) )
STATIC_ROOT = os.path.join(BASE_DIR, "static_media/") STATIC_ROOT = BASE_DIR / "static_media/"
REST_FRAMEWORK = { REST_FRAMEWORK = {
"PAGE_SIZE": 10, "PAGE_SIZE": 10,
@ -180,5 +180,5 @@ JENKINS_TOKEN = "sipwise_jenkins_ci"
CELERY_TASK_SERIALIZER = "json" CELERY_TASK_SERIALIZER = "json"
CELERY_RESULT_SERIALIZER = "json" CELERY_RESULT_SERIALIZER = "json"
CELERY_ACCEPT_CONTENT = ["json"] CELERY_ACCEPT_CONTENT = ["application/json"]
CELERY_RESULT_BACKEND = "django-db" CELERY_RESULT_BACKEND = "django-db"

4
repoapi/settings/dev.py
Unescape View File

@ -1,4 +1,4 @@
# Copyright (C) 2015 The Sipwise Team - http://sipwise.com # Copyright (C) 2015-2022 The Sipwise Team - http://sipwise.com
# #
# This program is free software: you can redistribute it and/or modify it # This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU General Public License as published by the Free # under the terms of the GNU General Public License as published by the Free
@ -29,7 +29,7 @@ LOGGING["loggers"]["repoapi"]["level"] = os.getenv( # noqa
BROKER_BACKEND = "amqp" BROKER_BACKEND = "amqp"
CELERY_TASK_ALWAYS_EAGER = False CELERY_TASK_ALWAYS_EAGER = False
CELERY_BROKER_URL = "amqp://guest:guest@rabbit" CELERY_BROKER_URL = "amqp://guest:guest@rabbit"
JBI_BASEDIR = os.path.join(BASE_DIR, "jbi_files") # noqa JBI_BASEDIR = BASE_DIR / "jbi_files" # noqa
# Enable access when not accessing from localhost: # Enable access when not accessing from localhost:
ALLOWED_HOSTS = [ ALLOWED_HOSTS = [

22
repoapi/settings/prod.py
Unescape View File

@ -16,8 +16,7 @@
# Build paths inside the project like this: join(BASE_DIR, ...) # Build paths inside the project like this: join(BASE_DIR, ...)
import os import os
from configparser import RawConfigParser from configparser import RawConfigParser
from os.path import dirname from pathlib import Path
from os.path import join
from urllib.parse import urlparse from urllib.parse import urlparse
from celery.schedules import crontab from celery.schedules import crontab
@ -26,10 +25,11 @@ from .common import * # noqa
# pylint: disable=W0401,W0614 # pylint: disable=W0401,W0614
BASE_DIR = dirname(dirname(dirname(os.path.abspath(__file__)))) # Build paths inside the project like this: BASE_DIR / 'subdir'.
BASE_DIR = Path(__file__).resolve(strict=True).parent.parent.parent
VAR_DIR = "/var/lib/repoapi" VAR_DIR = Path("/var/lib/repoapi")
if not os.path.exists(VAR_DIR): if not VAR_DIR.exists():
VAR_DIR = BASE_DIR VAR_DIR = BASE_DIR
# Quick-start development settings - unsuitable for production # Quick-start development settings - unsuitable for production
@ -37,7 +37,7 @@ if not os.path.exists(VAR_DIR):
# SECURITY WARNING: keep the secret key used in production secret! # SECURITY WARNING: keep the secret key used in production secret!
# read it from external file # read it from external file
SECRET_KEY = open(join(VAR_DIR, ".secret_key")).read().strip() SECRET_KEY = (VAR_DIR / ".secret_key").read_text().strip()
# SECURITY WARNING: don't run with debug turned on in production! # SECURITY WARNING: don't run with debug turned on in production!
DEBUG = False DEBUG = False
@ -49,7 +49,7 @@ LOGGING["loggers"]["repoapi"]["level"] = os.getenv( # noqa
) # noqa ) # noqa
server_config = RawConfigParser() server_config = RawConfigParser()
server_config.read(join(VAR_DIR, "server.ini")) server_config.read(VAR_DIR / "server.ini")
JENKINS_URL = server_config.get("jenkins", "URL") JENKINS_URL = server_config.get("jenkins", "URL")
JENKINS_HTTP_USER = server_config.get("jenkins", "HTTP_USER") JENKINS_HTTP_USER = server_config.get("jenkins", "HTTP_USER")
@ -89,12 +89,14 @@ STATICFILES_STORAGE = (
"django.contrib.staticfiles.storage.ManifestStaticFilesStorage" "django.contrib.staticfiles.storage.ManifestStaticFilesStorage"
) )
GITWEB_URL = "https://git.mgm.sipwise.com/gitweb/?p={}.git;a=commit;h={}" GITWEB_URL = "https://git.mgm.sipwise.com/gitweb/?p={}.git;a=commit;h={}"
WORKFRONT_CREDENTIALS = join(BASE_DIR, "/etc/jenkins_jobs/workfront.ini") WORKFRONT_CREDENTIALS = BASE_DIR / "/etc/jenkins_jobs/workfront.ini"
WORKFRONT_NOTE = True WORKFRONT_NOTE = True
# build app # build app
BUILD_KEY_AUTH = True BUILD_KEY_AUTH = True
BUILD_REPOS_SCRIPTS_CONFIG_DIR = "/usr/share/sipwise-repos-scripts/config" BUILD_REPOS_SCRIPTS_CONFIG_DIR = Path(
"/usr/share/sipwise-repos-scripts/config"
)
# celery # celery
CELERY_BROKER_URL = server_config.get("server", "BROKER_URL") CELERY_BROKER_URL = server_config.get("server", "BROKER_URL")
@ -113,7 +115,7 @@ CELERY_BEAT_SCHEDULE = {
} }
CELERY_TIMEZONE = "UTC" CELERY_TIMEZONE = "UTC"
JBI_BASEDIR = join(VAR_DIR, "jbi_files") JBI_BASEDIR = VAR_DIR / "jbi_files"
JBI_ARTIFACT_JOBS = [ JBI_ARTIFACT_JOBS = [
"release-tools-runner", "release-tools-runner",
] ]

21
repoapi/settings/test.py
Unescape View File

@ -1,4 +1,4 @@
# Copyright (C) 2015 The Sipwise Team - http://sipwise.com # Copyright (C) 2015-2022 The Sipwise Team - http://sipwise.com
# #
# This program is free software: you can redistribute it and/or modify it # This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU General Public License as published by the Free # under the terms of the GNU General Public License as published by the Free
@ -13,18 +13,17 @@
# You should have received a copy of the GNU General Public License along # You should have received a copy of the GNU General Public License along
# with this program. If not, see <http://www.gnu.org/licenses/>. # with this program. If not, see <http://www.gnu.org/licenses/>.
# #
# Build paths inside the project like this: join(BASE_DIR, ...)
import os import os
from os.path import dirname from pathlib import Path
from os.path import join
from .common import * # noqa from .common import * # noqa
# pylint: disable=W0401,W0614 # pylint: disable=W0401,W0614
BASE_DIR = dirname(dirname(dirname(os.path.abspath(__file__)))) # Build paths inside the project like this: BASE_DIR / 'subdir'.
BASE_DIR = Path(__file__).resolve(strict=True).parent.parent.parent
os.environ.setdefault("RESULTS", "/tmp") os.environ.setdefault("RESULTS", "/tmp")
RESULTS_DIR = os.environ["RESULTS"] RESULTS_DIR = Path(os.environ["RESULTS"])
# Quick-start development settings - unsuitable for production # Quick-start development settings - unsuitable for production
# See https://docs.djangoproject.com/en/1.8/howto/deployment/checklist/ # See https://docs.djangoproject.com/en/1.8/howto/deployment/checklist/
@ -43,7 +42,7 @@ ALLOWED_HOSTS = []
DATABASES = { DATABASES = {
"default": { "default": {
"ENGINE": "django.db.backends.sqlite3", "ENGINE": "django.db.backends.sqlite3",
"NAME": join(BASE_DIR, "db.sqlite3"), "NAME": BASE_DIR / "db.sqlite3",
} }
} }
@ -57,7 +56,7 @@ GERRIT_URL = "https://gerrit.local/{}"
GERRIT_REST_HTTP_USER = "jenkins" GERRIT_REST_HTTP_USER = "jenkins"
GERRIT_REST_HTTP_PASSWD = "verysecrethttppasswd" GERRIT_REST_HTTP_PASSWD = "verysecrethttppasswd"
GITWEB_URL = "https://git.local/gitweb/?p={}.git;a=commit;h={}" GITWEB_URL = "https://git.local/gitweb/?p={}.git;a=commit;h={}"
WORKFRONT_CREDENTIALS = join(BASE_DIR, ".workfront.ini") WORKFRONT_CREDENTIALS = BASE_DIR / ".workfront.ini"
WORKFRONT_NOTE = True WORKFRONT_NOTE = True
DOCKER_REGISTRY_URL = "https://localhost:5000/v2/{}" DOCKER_REGISTRY_URL = "https://localhost:5000/v2/{}"
# fake info # fake info
@ -96,13 +95,15 @@ RELEASE_DASHBOARD_DOCKER_IMAGES = {
# build app # build app
BUILD_KEY_AUTH = True BUILD_KEY_AUTH = True
BUILD_REPOS_SCRIPTS_CONFIG_DIR = join(BASE_DIR, "build", "fixtures", "config") BUILD_REPOS_SCRIPTS_CONFIG_DIR = BASE_DIR.joinpath(
"build", "fixtures", "config"
)
# celery # celery
CELERY_BROKER_URL = "memory://localhost/" CELERY_BROKER_URL = "memory://localhost/"
CELERY_TASK_ALWAYS_EAGER = True CELERY_TASK_ALWAYS_EAGER = True
CELERY_TASK_EAGER_PROPAGATES = True CELERY_TASK_EAGER_PROPAGATES = True
JBI_BASEDIR = join(RESULTS_DIR, "jbi_files") JBI_BASEDIR = RESULTS_DIR / "jbi_files"
JBI_ARTIFACT_JOBS = [ JBI_ARTIFACT_JOBS = [
"fake-release-tools-runner", "fake-release-tools-runner",
] ]

5
repoapi/tasks.py
Unescape View File

@ -1,4 +1,4 @@
# Copyright (C) 2016-2020 The Sipwise Team - http://sipwise.com # Copyright (C) 2016-2022 The Sipwise Team - http://sipwise.com
# #
# This program is free software: you can redistribute it and/or modify it # This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU General Public License as published by the Free # under the terms of the GNU General Public License as published by the Free
@ -14,7 +14,6 @@
# with this program. If not, see <http://www.gnu.org/licenses/>. # with this program. If not, see <http://www.gnu.org/licenses/>.
import json import json
from datetime import timedelta from datetime import timedelta
from pathlib import Path
import structlog import structlog
from celery import shared_task from celery import shared_task
@ -54,7 +53,7 @@ def jenkins_remove_project(self, jbi_id):
@shared_task(ignore_result=True) @shared_task(ignore_result=True)
def jbi_get_artifact(jbi_id, jobname, buildnumber, artifact_info): def jbi_get_artifact(jbi_id, jobname, buildnumber, artifact_info):
path = Path(jenkins_get_artifact(jobname, buildnumber, artifact_info)) path = jenkins_get_artifact(jobname, buildnumber, artifact_info)
if path.name == settings.HOTFIX_ARTIFACT: if path.name == settings.HOTFIX_ARTIFACT:
jbi_parse_hotfix.delay(jbi_id, str(path)) jbi_parse_hotfix.delay(jbi_id, str(path))

23
repoapi/test/base.py
Unescape View File

@ -1,4 +1,4 @@
# Copyright (C) 2017-2020 The Sipwise Team - http://sipwise.com # Copyright (C) 2017-2022 The Sipwise Team - http://sipwise.com
# #
# This program is free software: you can redistribute it and/or modify it # This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU General Public License as published by the Free # under the terms of the GNU General Public License as published by the Free
@ -13,8 +13,8 @@
# You should have received a copy of the GNU General Public License along # You should have received a copy of the GNU General Public License along
# with this program. If not, see <http://www.gnu.org/licenses/>. # with this program. If not, see <http://www.gnu.org/licenses/>.
import os import os
from distutils.dir_util import mkpath import shutil
from distutils.dir_util import remove_tree from pathlib import Path
from tempfile import mkdtemp from tempfile import mkdtemp
from django.test import override_settings from django.test import override_settings
@ -23,21 +23,24 @@ from rest_framework.test import APITestCase
from rest_framework_api_key.helpers import generate_key from rest_framework_api_key.helpers import generate_key
from rest_framework_api_key.models import APIKey from rest_framework_api_key.models import APIKey
JBI_BASEDIR = mkdtemp(dir=os.environ.get("RESULTS")) JBI_BASEDIR = Path(mkdtemp(dir=os.environ.get("RESULTS")))
@override_settings(DEBUG=True, JBI_BASEDIR=JBI_BASEDIR) @override_settings(DEBUG=True, JBI_BASEDIR=JBI_BASEDIR)
class BaseTest(TestCase): class BaseTest(TestCase):
def setUp(self, *args, **kwargs): @classmethod
def setUpTestData(cls):
from repoapi.conf import settings from repoapi.conf import settings
mkpath(settings.JBI_BASEDIR, verbose=True) cls.path = Path(settings.JBI_BASEDIR)
def tearDown(self, *args, **kwargs): def setUp(self, *args, **kwargs):
from repoapi.conf import settings super(BaseTest, self).setUp()
self.path.mkdir(parents=True, exist_ok=True)
if os.path.exists(settings.JBI_BASEDIR): def tearDown(self, *args, **kwargs):
remove_tree(settings.JBI_BASEDIR, verbose=True) if self.path.exists():
shutil.rmtree(self.path)
class APIAuthenticatedTestCase(BaseTest, APITestCase): class APIAuthenticatedTestCase(BaseTest, APITestCase):

5
repoapi/test/test_basic_model.py
Unescape View File

@ -12,7 +12,6 @@
# #
# You should have received a copy of the GNU General Public License along # You should have received a copy of the GNU General Public License along
# with this program. If not, see <http://www.gnu.org/licenses/>. # with this program. If not, see <http://www.gnu.org/licenses/>.
from os.path import join
from unittest.mock import mock_open from unittest.mock import mock_open
from unittest.mock import patch from unittest.mock import patch
@ -22,7 +21,7 @@ from .base import BaseTest
from repoapi.conf import settings from repoapi.conf import settings
from repoapi.models import JenkinsBuildInfo from repoapi.models import JenkinsBuildInfo
FIXTURES_PATH = join(settings.BASE_DIR, "repoapi", "fixtures", "jbi_files") FIXTURES_PATH = settings.BASE_DIR.joinpath("repoapi", "fixtures", "jbi_files")
JBI_HOST = "https://%s/job/fake-gerrit/" JBI_HOST = "https://%s/job/fake-gerrit/"
ARTIFACTS_JSON = """{ ARTIFACTS_JSON = """{
"artifacts": [ "artifacts": [
@ -107,7 +106,7 @@ class JenkinsBuildInfoProperties(BaseTest):
self.jbi = JenkinsBuildInfo.objects.get(id=1) self.jbi = JenkinsBuildInfo.objects.get(id=1)
def test_build_path(self): def test_build_path(self):
self.assertRegex(self.jbi.build_path, "^.+/fake-source/1$") self.assertRegex(str(self.jbi.build_path), "^.+/fake-source/1$")
@patch("builtins.open", mock_open(read_data="{}")) @patch("builtins.open", mock_open(read_data="{}"))
def test_build_info(self): def test_build_info(self):

47
repoapi/test/test_jbi_info.py
Unescape View File

@ -12,7 +12,6 @@
# #
# You should have received a copy of the GNU General Public License along # You should have received a copy of the GNU General Public License along
# with this program. If not, see <http://www.gnu.org/licenses/>. # with this program. If not, see <http://www.gnu.org/licenses/>.
import os
from unittest.mock import call from unittest.mock import call
from unittest.mock import mock_open from unittest.mock import mock_open
from unittest.mock import patch from unittest.mock import patch
@ -62,21 +61,17 @@ class TestJBICelery(BaseTest):
param = self.get_defaults() param = self.get_defaults()
param["jobname"] = "fake-me" param["jobname"] = "fake-me"
jbi = JenkinsBuildInfo.objects.create(**param) jbi = JenkinsBuildInfo.objects.create(**param)
base_path = os.path.join( base_path = self.path.joinpath(jbi.jobname, str(jbi.buildnumber))
settings.JBI_BASEDIR, jbi.jobname, str(jbi.buildnumber) self.assertTrue(base_path.is_dir(), base_path)
)
self.assertTrue(os.path.isdir(base_path), base_path)
@patch("builtins.open", mock_open(read_data=artifacts_json)) @patch("builtins.open", mock_open(read_data=artifacts_json))
@patch("repoapi.utils.dlfile") @patch("repoapi.utils.dlfile")
def test_jbi_console(self, dlfile): def test_jbi_console(self, dlfile):
param = self.get_defaults() param = self.get_defaults()
jbi = JenkinsBuildInfo.objects.create(**param) jbi = JenkinsBuildInfo.objects.create(**param)
base_path = os.path.join( base_path = self.path.joinpath(jbi.jobname, str(jbi.buildnumber))
settings.JBI_BASEDIR, jbi.jobname, str(jbi.buildnumber)
)
path = os.path.join(base_path, "console.txt") path = base_path.joinpath("console.txt")
url = JBI_CONSOLE_URL.format( url = JBI_CONSOLE_URL.format(
settings.JENKINS_URL, jbi.jobname, jbi.buildnumber settings.JENKINS_URL, jbi.jobname, jbi.buildnumber
) )
@ -87,8 +82,8 @@ class TestJBICelery(BaseTest):
jbi.buildnumber, jbi.buildnumber,
"builddeps.list", "builddeps.list",
) )
artifact_base_path = os.path.join(base_path, "artifact") artifact_base_path = base_path.joinpath("artifact")
path = os.path.join(artifact_base_path, "builddeps.list") path = artifact_base_path.joinpath("builddeps.list")
self.assertNotIn(call(url, path), dlfile.call_args_list) self.assertNotIn(call(url, path), dlfile.call_args_list)
@patch("builtins.open", mock_open(read_data=artifacts_json)) @patch("builtins.open", mock_open(read_data=artifacts_json))
@ -96,13 +91,11 @@ class TestJBICelery(BaseTest):
def test_jbi_buildinfo(self, dlfile): def test_jbi_buildinfo(self, dlfile):
param = self.get_defaults() param = self.get_defaults()
jbi = JenkinsBuildInfo.objects.create(**param) jbi = JenkinsBuildInfo.objects.create(**param)
base_path = os.path.join( base_path = self.path.joinpath(jbi.jobname, str(jbi.buildnumber))
settings.JBI_BASEDIR, jbi.jobname, str(jbi.buildnumber)
)
url = JBI_BUILD_URL.format( url = JBI_BUILD_URL.format(
settings.JENKINS_URL, jbi.jobname, jbi.buildnumber settings.JENKINS_URL, jbi.jobname, jbi.buildnumber
) )
path = os.path.join(base_path, "build.json") path = base_path.joinpath("build.json")
dlfile.assert_any_call(url, path) dlfile.assert_any_call(url, path)
url = JBI_ARTIFACT_URL.format( url = JBI_ARTIFACT_URL.format(
settings.JENKINS_URL, settings.JENKINS_URL,
@ -110,8 +103,8 @@ class TestJBICelery(BaseTest):
jbi.buildnumber, jbi.buildnumber,
"builddeps.list", "builddeps.list",
) )
artifact_base_path = os.path.join(base_path, "artifact") artifact_base_path = base_path.joinpath("artifact")
path = os.path.join(artifact_base_path, "builddeps.list") path = artifact_base_path.joinpath("builddeps.list")
self.assertNotIn(call(url, path), dlfile.call_args_list) self.assertNotIn(call(url, path), dlfile.call_args_list)
@patch("builtins.open", mock_open(read_data=artifacts_json)) @patch("builtins.open", mock_open(read_data=artifacts_json))
@ -120,17 +113,15 @@ class TestJBICelery(BaseTest):
param = self.get_defaults() param = self.get_defaults()
param["jobname"] = "fake-release-tools-runner" param["jobname"] = "fake-release-tools-runner"
jbi = JenkinsBuildInfo.objects.create(**param) jbi = JenkinsBuildInfo.objects.create(**param)
base_path = os.path.join( base_path = self.path.joinpath(jbi.jobname, str(jbi.buildnumber))
settings.JBI_BASEDIR, jbi.jobname, str(jbi.buildnumber)
)
url = JBI_ARTIFACT_URL.format( url = JBI_ARTIFACT_URL.format(
settings.JENKINS_URL, settings.JENKINS_URL,
jbi.jobname, jbi.jobname,
jbi.buildnumber, jbi.buildnumber,
"builddeps.list", "builddeps.list",
) )
artifact_base_path = os.path.join(base_path, "artifact") artifact_base_path = base_path.joinpath("artifact")
path = os.path.join(artifact_base_path, "builddeps.list") path = artifact_base_path.joinpath("builddeps.list")
dlfile.assert_any_call(url, path) dlfile.assert_any_call(url, path)
@patch("builtins.open", mock_open(read_data=artifacts_json)) @patch("builtins.open", mock_open(read_data=artifacts_json))
@ -138,13 +129,11 @@ class TestJBICelery(BaseTest):
def test_jbi_envVars(self, dlfile): def test_jbi_envVars(self, dlfile):
param = self.get_defaults() param = self.get_defaults()
jbi = JenkinsBuildInfo.objects.create(**param) jbi = JenkinsBuildInfo.objects.create(**param)
base_path = os.path.join( base_path = self.path.joinpath(jbi.jobname, str(jbi.buildnumber))
settings.JBI_BASEDIR, jbi.jobname, str(jbi.buildnumber)
)
url = JBI_ENVVARS_URL.format( url = JBI_ENVVARS_URL.format(
settings.JENKINS_URL, jbi.jobname, jbi.buildnumber settings.JENKINS_URL, jbi.jobname, jbi.buildnumber
) )
path = os.path.join(base_path, "envVars.json") path = base_path.joinpath("envVars.json")
dlfile.assert_any_call(url, path) dlfile.assert_any_call(url, path)
@ -162,13 +151,11 @@ class TestJBIReleaseChangedCelery(BaseTest):
"/check-ngcp-release-changed", "/check-ngcp-release-changed",
} }
jbi = JenkinsBuildInfo.objects.create(**param) jbi = JenkinsBuildInfo.objects.create(**param)
base_path = os.path.join( base_path = self.path.joinpath(jbi.jobname, str(jbi.buildnumber))
settings.JBI_BASEDIR, jbi.jobname, str(jbi.buildnumber)
)
url = JBI_ENVVARS_URL.format( url = JBI_ENVVARS_URL.format(
settings.JENKINS_URL, jbi.jobname, jbi.buildnumber settings.JENKINS_URL, jbi.jobname, jbi.buildnumber
) )
path = os.path.join(base_path, "envVars.json") path = base_path.joinpath("envVars.json")
dlfile.assert_any_call(url, path) dlfile.assert_any_call(url, path)
app.send_task.assert_called_once_with( app.send_task.assert_called_once_with(
"release_changed.tasks.process_result", args=[jbi.id, path] "release_changed.tasks.process_result", args=[jbi.id, path]

5
repoapi/test/test_tasks.py
Unescape View File

@ -1,4 +1,4 @@
# Copyright (C) 2017-2020 The Sipwise Team - http://sipwise.com # Copyright (C) 2017-2022 The Sipwise Team - http://sipwise.com
# #
# This program is free software: you can redistribute it and/or modify it # This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU General Public License as published by the Free # under the terms of the GNU General Public License as published by the Free
@ -14,7 +14,6 @@
# with this program. If not, see <http://www.gnu.org/licenses/>. # with this program. If not, see <http://www.gnu.org/licenses/>.
from copy import deepcopy from copy import deepcopy
from datetime import datetime from datetime import datetime
from os.path import join
from unittest.mock import patch from unittest.mock import patch
from django.test import override_settings from django.test import override_settings
@ -26,7 +25,7 @@ from repoapi.models import GerritRepoInfo
from repoapi.models import JenkinsBuildInfo from repoapi.models import JenkinsBuildInfo
from repoapi.test.base import BaseTest from repoapi.test.base import BaseTest
FIXTURES_PATH = join(settings.BASE_DIR, "repoapi", "fixtures", "jbi_files") FIXTURES_PATH = settings.BASE_DIR.joinpath("repoapi", "fixtures", "jbi_files")
class TasksTestCase(BaseTest): class TasksTestCase(BaseTest):

23
repoapi/utils.py
Unescape View File

@ -1,4 +1,4 @@
# Copyright (C) 2015-2020 The Sipwise Team - http://sipwise.com # Copyright (C) 2015-2022 The Sipwise Team - http://sipwise.com
# #
# This program is free software: you can redistribute it and/or modify it # This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU General Public License as published by the Free # under the terms of the GNU General Public License as published by the Free
@ -12,10 +12,9 @@
# #
# You should have received a copy of the GNU General Public License along # You should have received a copy of the GNU General Public License along
# with this program. If not, see <http://www.gnu.org/licenses/>. # with this program. If not, see <http://www.gnu.org/licenses/>.
import os
import re import re
import subprocess import subprocess
from distutils.dir_util import mkpath from pathlib import Path
import requests import requests
import structlog import structlog
@ -53,7 +52,7 @@ def get_jenkins_response(url):
return response return response
def dlfile(url, path): def dlfile(url, path: Path):
log = logger.bind( log = logger.bind(
url=url, url=url,
path=path, path=path,
@ -123,9 +122,9 @@ def jenkins_remove_project_ppa(repo, source):
open_jenkins_url(url) open_jenkins_url(url)
def _jenkins_get(url, base_path, filename): def _jenkins_get(url, base_path: Path, filename) -> Path:
mkpath(base_path) base_path.mkdir(parents=True, exist_ok=True)
path = os.path.join(base_path, filename) path = base_path.joinpath(filename)
log = logger.bind( log = logger.bind(
base_path=base_path, base_path=base_path,
filename=filename, filename=filename,
@ -138,19 +137,19 @@ def _jenkins_get(url, base_path, filename):
def jenkins_get_console(jobname, buildnumber): def jenkins_get_console(jobname, buildnumber):
url = JBI_CONSOLE_URL.format(settings.JENKINS_URL, jobname, buildnumber) url = JBI_CONSOLE_URL.format(settings.JENKINS_URL, jobname, buildnumber)
base_path = os.path.join(settings.JBI_BASEDIR, jobname, str(buildnumber)) base_path = settings.JBI_BASEDIR.joinpath(jobname, str(buildnumber))
return _jenkins_get(url, base_path, "console.txt") return _jenkins_get(url, base_path, "console.txt")
def jenkins_get_build(jobname, buildnumber): def jenkins_get_build(jobname, buildnumber):
url = JBI_BUILD_URL.format(settings.JENKINS_URL, jobname, buildnumber) url = JBI_BUILD_URL.format(settings.JENKINS_URL, jobname, buildnumber)
base_path = os.path.join(settings.JBI_BASEDIR, jobname, str(buildnumber)) base_path = settings.JBI_BASEDIR.joinpath(jobname, str(buildnumber))
return _jenkins_get(url, base_path, "build.json") return _jenkins_get(url, base_path, "build.json")
def jenkins_get_env(jobname, buildnumber): def jenkins_get_env(jobname, buildnumber):
url = JBI_ENVVARS_URL.format(settings.JENKINS_URL, jobname, buildnumber) url = JBI_ENVVARS_URL.format(settings.JENKINS_URL, jobname, buildnumber)
base_path = os.path.join(settings.JBI_BASEDIR, jobname, str(buildnumber)) base_path = settings.JBI_BASEDIR.joinpath(jobname, str(buildnumber))
return _jenkins_get(url, base_path, "envVars.json") return _jenkins_get(url, base_path, "envVars.json")
@ -161,8 +160,8 @@ def jenkins_get_artifact(jobname, buildnumber, artifact_info):
buildnumber, buildnumber,
artifact_info["relativePath"], artifact_info["relativePath"],
) )
base_path = os.path.join( base_path = settings.JBI_BASEDIR.joinpath(
settings.JBI_BASEDIR, jobname, str(buildnumber), "artifact" jobname, str(buildnumber), "artifact"
) )
return _jenkins_get(url, base_path, artifact_info["fileName"]) return _jenkins_get(url, base_path, artifact_info["fileName"])

Write Preview
Loading…
Cancel
Save