diff --git a/build/utils.py b/build/utils.py
index fca63be..91901b9 100644
--- a/build/utils.py
+++ b/build/utils.py
@@ -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
# 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
# with this program. If not, see .
import logging
-import os
import re
import urllib
+from os import walk
from pathlib import Path
from uuid import uuid4
@@ -180,15 +180,11 @@ class ReleaseConfig(object):
def supported_releases(cls):
skip_files = ["{}.yml".format(x) for x in settings.BUILD_RELEASES_SKIP]
res = []
- for root, dirs, files in os.walk(
- settings.BUILD_REPOS_SCRIPTS_CONFIG_DIR
- ):
+ for root, dirs, files in walk(settings.BUILD_REPOS_SCRIPTS_CONFIG_DIR):
if "trunk.yml" in files:
files.remove("trunk.yml")
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"]:
res.append(dist)
@@ -217,8 +213,8 @@ class ReleaseConfig(object):
if filename is None:
filename = name
self.config_file = "{}.yml".format(filename)
- self.config_path = os.path.join(
- settings.BUILD_REPOS_SCRIPTS_CONFIG_DIR, self.config_file
+ self.config_path = (
+ settings.BUILD_REPOS_SCRIPTS_CONFIG_DIR / self.config_file
)
self.config = self.load_config(self.config_path)
try:
diff --git a/release_changed/test_tasks.py b/release_changed/test_tasks.py
index 4e1469c..bc8bbb5 100644
--- a/release_changed/test_tasks.py
+++ b/release_changed/test_tasks.py
@@ -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
# 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
# with this program. If not, see .
-from os.path import join
from unittest.mock import mock_open
from unittest.mock import patch
@@ -22,13 +21,11 @@ from .models import ReleaseChanged
from repoapi.models import JenkinsBuildInfo
from repoapi.test.base import BaseTest
-FIXTURES_PATH = join(settings.BASE_DIR, "release_changed", "fixtures")
-FILE_PATH = join(FIXTURES_PATH, "test_envVars.json")
-FILE_PATH_DONE = join(FIXTURES_PATH, "test_envVars_done.json")
-with open(FILE_PATH) as file:
- DATA = file.read()
-with open(FILE_PATH_DONE) as file:
- DATA_DONE = file.read()
+FIXTURES_PATH = settings.BASE_DIR.joinpath("release_changed", "fixtures")
+FILE_PATH = FIXTURES_PATH / "test_envVars.json"
+FILE_PATH_DONE = FIXTURES_PATH / "test_envVars_done.json"
+DATA = FILE_PATH.read_text()
+DATA_DONE = FILE_PATH_DONE.read_text()
class TasksTestCase(BaseTest):
@@ -37,7 +34,7 @@ class TasksTestCase(BaseTest):
def test_process_create(self):
jbi = JenkinsBuildInfo.objects.get(pk=1)
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(
label="base", version="mr8.5.1", vmtype="CE"
)
@@ -50,7 +47,7 @@ class TasksTestCase(BaseTest):
r_id = r.id
jbi = JenkinsBuildInfo.objects.get(pk=1)
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(
label="base", version="mr8.5.1", vmtype="CE"
)
@@ -63,7 +60,7 @@ class TasksTestCase(BaseTest):
)
jbi = JenkinsBuildInfo.objects.get(pk=1)
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(
label="base", version="mr8.5.1", vmtype="CE"
)
@@ -72,7 +69,7 @@ class TasksTestCase(BaseTest):
def test_process_done_no_obj(self):
jbi = JenkinsBuildInfo.objects.get(pk=1)
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(
label="base", version="mr8.5.1", vmtype="CE"
)
diff --git a/repoapi/models/jbi.py b/repoapi/models/jbi.py
index 8a46ca8..24cd91a 100644
--- a/repoapi/models/jbi.py
+++ b/repoapi/models/jbi.py
@@ -18,7 +18,6 @@ import re
from collections import OrderedDict
from datetime import datetime
from datetime import timedelta
-from os.path import join
from urllib.parse import urlparse
import structlog
@@ -252,11 +251,13 @@ class JenkinsBuildInfo(models.Model):
@property
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
def build_info(self):
- path = join(self.build_path, "build.json")
+ path = self.build_path.joinpath("build.json")
try:
with open(path, "r") as data_file:
data = json.load(data_file)
@@ -278,8 +279,8 @@ class JenkinsBuildInfo(models.Model):
return getattr(self, "_source")
except AttributeError:
pass
- path = join(self.build_path, "artifact")
- dscs = glob.glob(join(path, "*.dsc"))
+ path = self.build_path.joinpath("artifact")
+ dscs = glob.glob(str(path.joinpath("*.dsc")))
if len(dscs) != 1:
logger.error("more than one dsc file on artifact dir", path=path)
return None
diff --git a/repoapi/settings/common.py b/repoapi/settings/common.py
index 8139f5a..c837bd9 100644
--- a/repoapi/settings/common.py
+++ b/repoapi/settings/common.py
@@ -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
# 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
# with this program. If not, see .
#
-# Build paths inside the project like this: os.path.join(BASE_DIR, ...)
import os
-from os.path import dirname
+from pathlib import Path
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 = [
"django.contrib.admin",
@@ -105,7 +105,7 @@ STATICFILES_FINDERS = (
"django_assets.finders.AssetsFinder",
)
-STATIC_ROOT = os.path.join(BASE_DIR, "static_media/")
+STATIC_ROOT = BASE_DIR / "static_media/"
REST_FRAMEWORK = {
"PAGE_SIZE": 10,
@@ -180,5 +180,5 @@ JENKINS_TOKEN = "sipwise_jenkins_ci"
CELERY_TASK_SERIALIZER = "json"
CELERY_RESULT_SERIALIZER = "json"
-CELERY_ACCEPT_CONTENT = ["json"]
+CELERY_ACCEPT_CONTENT = ["application/json"]
CELERY_RESULT_BACKEND = "django-db"
diff --git a/repoapi/settings/dev.py b/repoapi/settings/dev.py
index 0c72d9e..8f8a233 100644
--- a/repoapi/settings/dev.py
+++ b/repoapi/settings/dev.py
@@ -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
# 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"
CELERY_TASK_ALWAYS_EAGER = False
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:
ALLOWED_HOSTS = [
diff --git a/repoapi/settings/prod.py b/repoapi/settings/prod.py
index 4cffa24..98bd238 100644
--- a/repoapi/settings/prod.py
+++ b/repoapi/settings/prod.py
@@ -16,8 +16,7 @@
# Build paths inside the project like this: join(BASE_DIR, ...)
import os
from configparser import RawConfigParser
-from os.path import dirname
-from os.path import join
+from pathlib import Path
from urllib.parse import urlparse
from celery.schedules import crontab
@@ -26,10 +25,11 @@ from .common import * # noqa
# 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"
-if not os.path.exists(VAR_DIR):
+VAR_DIR = Path("/var/lib/repoapi")
+if not VAR_DIR.exists():
VAR_DIR = BASE_DIR
# 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!
# 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!
DEBUG = False
@@ -49,7 +49,7 @@ LOGGING["loggers"]["repoapi"]["level"] = os.getenv( # noqa
) # noqa
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_HTTP_USER = server_config.get("jenkins", "HTTP_USER")
@@ -89,12 +89,14 @@ STATICFILES_STORAGE = (
"django.contrib.staticfiles.storage.ManifestStaticFilesStorage"
)
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
# build app
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_BROKER_URL = server_config.get("server", "BROKER_URL")
@@ -113,7 +115,7 @@ CELERY_BEAT_SCHEDULE = {
}
CELERY_TIMEZONE = "UTC"
-JBI_BASEDIR = join(VAR_DIR, "jbi_files")
+JBI_BASEDIR = VAR_DIR / "jbi_files"
JBI_ARTIFACT_JOBS = [
"release-tools-runner",
]
diff --git a/repoapi/settings/test.py b/repoapi/settings/test.py
index 63e5158..ec980f0 100644
--- a/repoapi/settings/test.py
+++ b/repoapi/settings/test.py
@@ -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
# 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
# with this program. If not, see .
#
-# Build paths inside the project like this: join(BASE_DIR, ...)
import os
-from os.path import dirname
-from os.path import join
+from pathlib import Path
from .common import * # noqa
# 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")
-RESULTS_DIR = os.environ["RESULTS"]
+RESULTS_DIR = Path(os.environ["RESULTS"])
# Quick-start development settings - unsuitable for production
# See https://docs.djangoproject.com/en/1.8/howto/deployment/checklist/
@@ -43,7 +42,7 @@ ALLOWED_HOSTS = []
DATABASES = {
"default": {
"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_PASSWD = "verysecrethttppasswd"
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
DOCKER_REGISTRY_URL = "https://localhost:5000/v2/{}"
# fake info
@@ -96,13 +95,15 @@ RELEASE_DASHBOARD_DOCKER_IMAGES = {
# build app
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_BROKER_URL = "memory://localhost/"
CELERY_TASK_ALWAYS_EAGER = True
CELERY_TASK_EAGER_PROPAGATES = True
-JBI_BASEDIR = join(RESULTS_DIR, "jbi_files")
+JBI_BASEDIR = RESULTS_DIR / "jbi_files"
JBI_ARTIFACT_JOBS = [
"fake-release-tools-runner",
]
diff --git a/repoapi/tasks.py b/repoapi/tasks.py
index f1166d0..03d1221 100644
--- a/repoapi/tasks.py
+++ b/repoapi/tasks.py
@@ -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
# under the terms of the GNU General Public License as published by the Free
@@ -14,7 +14,6 @@
# with this program. If not, see .
import json
from datetime import timedelta
-from pathlib import Path
import structlog
from celery import shared_task
@@ -54,7 +53,7 @@ def jenkins_remove_project(self, jbi_id):
@shared_task(ignore_result=True)
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:
jbi_parse_hotfix.delay(jbi_id, str(path))
diff --git a/repoapi/test/base.py b/repoapi/test/base.py
index 5518dcb..0c66afc 100644
--- a/repoapi/test/base.py
+++ b/repoapi/test/base.py
@@ -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
# 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
# with this program. If not, see .
import os
-from distutils.dir_util import mkpath
-from distutils.dir_util import remove_tree
+import shutil
+from pathlib import Path
from tempfile import mkdtemp
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.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)
class BaseTest(TestCase):
- def setUp(self, *args, **kwargs):
+ @classmethod
+ def setUpTestData(cls):
from repoapi.conf import settings
- mkpath(settings.JBI_BASEDIR, verbose=True)
+ cls.path = Path(settings.JBI_BASEDIR)
- def tearDown(self, *args, **kwargs):
- from repoapi.conf import settings
+ def setUp(self, *args, **kwargs):
+ super(BaseTest, self).setUp()
+ self.path.mkdir(parents=True, exist_ok=True)
- if os.path.exists(settings.JBI_BASEDIR):
- remove_tree(settings.JBI_BASEDIR, verbose=True)
+ def tearDown(self, *args, **kwargs):
+ if self.path.exists():
+ shutil.rmtree(self.path)
class APIAuthenticatedTestCase(BaseTest, APITestCase):
diff --git a/repoapi/test/test_basic_model.py b/repoapi/test/test_basic_model.py
index 9cf7c34..480c579 100644
--- a/repoapi/test/test_basic_model.py
+++ b/repoapi/test/test_basic_model.py
@@ -12,7 +12,6 @@
#
# You should have received a copy of the GNU General Public License along
# with this program. If not, see .
-from os.path import join
from unittest.mock import mock_open
from unittest.mock import patch
@@ -22,7 +21,7 @@ from .base import BaseTest
from repoapi.conf import settings
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/"
ARTIFACTS_JSON = """{
"artifacts": [
@@ -107,7 +106,7 @@ class JenkinsBuildInfoProperties(BaseTest):
self.jbi = JenkinsBuildInfo.objects.get(id=1)
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="{}"))
def test_build_info(self):
diff --git a/repoapi/test/test_jbi_info.py b/repoapi/test/test_jbi_info.py
index 0dfc79b..be4f9be 100644
--- a/repoapi/test/test_jbi_info.py
+++ b/repoapi/test/test_jbi_info.py
@@ -12,7 +12,6 @@
#
# You should have received a copy of the GNU General Public License along
# with this program. If not, see .
-import os
from unittest.mock import call
from unittest.mock import mock_open
from unittest.mock import patch
@@ -62,21 +61,17 @@ class TestJBICelery(BaseTest):
param = self.get_defaults()
param["jobname"] = "fake-me"
jbi = JenkinsBuildInfo.objects.create(**param)
- base_path = os.path.join(
- settings.JBI_BASEDIR, jbi.jobname, str(jbi.buildnumber)
- )
- self.assertTrue(os.path.isdir(base_path), base_path)
+ base_path = self.path.joinpath(jbi.jobname, str(jbi.buildnumber))
+ self.assertTrue(base_path.is_dir(), base_path)
@patch("builtins.open", mock_open(read_data=artifacts_json))
@patch("repoapi.utils.dlfile")
def test_jbi_console(self, dlfile):
param = self.get_defaults()
jbi = JenkinsBuildInfo.objects.create(**param)
- base_path = os.path.join(
- settings.JBI_BASEDIR, jbi.jobname, str(jbi.buildnumber)
- )
+ base_path = self.path.joinpath(jbi.jobname, str(jbi.buildnumber))
- path = os.path.join(base_path, "console.txt")
+ path = base_path.joinpath("console.txt")
url = JBI_CONSOLE_URL.format(
settings.JENKINS_URL, jbi.jobname, jbi.buildnumber
)
@@ -87,8 +82,8 @@ class TestJBICelery(BaseTest):
jbi.buildnumber,
"builddeps.list",
)
- artifact_base_path = os.path.join(base_path, "artifact")
- path = os.path.join(artifact_base_path, "builddeps.list")
+ artifact_base_path = base_path.joinpath("artifact")
+ path = artifact_base_path.joinpath("builddeps.list")
self.assertNotIn(call(url, path), dlfile.call_args_list)
@patch("builtins.open", mock_open(read_data=artifacts_json))
@@ -96,13 +91,11 @@ class TestJBICelery(BaseTest):
def test_jbi_buildinfo(self, dlfile):
param = self.get_defaults()
jbi = JenkinsBuildInfo.objects.create(**param)
- base_path = os.path.join(
- settings.JBI_BASEDIR, jbi.jobname, str(jbi.buildnumber)
- )
+ base_path = self.path.joinpath(jbi.jobname, str(jbi.buildnumber))
url = JBI_BUILD_URL.format(
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)
url = JBI_ARTIFACT_URL.format(
settings.JENKINS_URL,
@@ -110,8 +103,8 @@ class TestJBICelery(BaseTest):
jbi.buildnumber,
"builddeps.list",
)
- artifact_base_path = os.path.join(base_path, "artifact")
- path = os.path.join(artifact_base_path, "builddeps.list")
+ artifact_base_path = base_path.joinpath("artifact")
+ path = artifact_base_path.joinpath("builddeps.list")
self.assertNotIn(call(url, path), dlfile.call_args_list)
@patch("builtins.open", mock_open(read_data=artifacts_json))
@@ -120,17 +113,15 @@ class TestJBICelery(BaseTest):
param = self.get_defaults()
param["jobname"] = "fake-release-tools-runner"
jbi = JenkinsBuildInfo.objects.create(**param)
- base_path = os.path.join(
- settings.JBI_BASEDIR, jbi.jobname, str(jbi.buildnumber)
- )
+ base_path = self.path.joinpath(jbi.jobname, str(jbi.buildnumber))
url = JBI_ARTIFACT_URL.format(
settings.JENKINS_URL,
jbi.jobname,
jbi.buildnumber,
"builddeps.list",
)
- artifact_base_path = os.path.join(base_path, "artifact")
- path = os.path.join(artifact_base_path, "builddeps.list")
+ artifact_base_path = base_path.joinpath("artifact")
+ path = artifact_base_path.joinpath("builddeps.list")
dlfile.assert_any_call(url, path)
@patch("builtins.open", mock_open(read_data=artifacts_json))
@@ -138,13 +129,11 @@ class TestJBICelery(BaseTest):
def test_jbi_envVars(self, dlfile):
param = self.get_defaults()
jbi = JenkinsBuildInfo.objects.create(**param)
- base_path = os.path.join(
- settings.JBI_BASEDIR, jbi.jobname, str(jbi.buildnumber)
- )
+ base_path = self.path.joinpath(jbi.jobname, str(jbi.buildnumber))
url = JBI_ENVVARS_URL.format(
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)
@@ -162,13 +151,11 @@ class TestJBIReleaseChangedCelery(BaseTest):
"/check-ngcp-release-changed",
}
jbi = JenkinsBuildInfo.objects.create(**param)
- base_path = os.path.join(
- settings.JBI_BASEDIR, jbi.jobname, str(jbi.buildnumber)
- )
+ base_path = self.path.joinpath(jbi.jobname, str(jbi.buildnumber))
url = JBI_ENVVARS_URL.format(
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)
app.send_task.assert_called_once_with(
"release_changed.tasks.process_result", args=[jbi.id, path]
diff --git a/repoapi/test/test_tasks.py b/repoapi/test/test_tasks.py
index f04f258..65d86a6 100644
--- a/repoapi/test/test_tasks.py
+++ b/repoapi/test/test_tasks.py
@@ -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
# under the terms of the GNU General Public License as published by the Free
@@ -14,7 +14,6 @@
# with this program. If not, see .
from copy import deepcopy
from datetime import datetime
-from os.path import join
from unittest.mock import patch
from django.test import override_settings
@@ -26,7 +25,7 @@ from repoapi.models import GerritRepoInfo
from repoapi.models import JenkinsBuildInfo
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):
diff --git a/repoapi/utils.py b/repoapi/utils.py
index 9e40034..7a95d71 100644
--- a/repoapi/utils.py
+++ b/repoapi/utils.py
@@ -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
# 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
# with this program. If not, see .
-import os
import re
import subprocess
-from distutils.dir_util import mkpath
+from pathlib import Path
import requests
import structlog
@@ -53,7 +52,7 @@ def get_jenkins_response(url):
return response
-def dlfile(url, path):
+def dlfile(url, path: Path):
log = logger.bind(
url=url,
path=path,
@@ -123,9 +122,9 @@ def jenkins_remove_project_ppa(repo, source):
open_jenkins_url(url)
-def _jenkins_get(url, base_path, filename):
- mkpath(base_path)
- path = os.path.join(base_path, filename)
+def _jenkins_get(url, base_path: Path, filename) -> Path:
+ base_path.mkdir(parents=True, exist_ok=True)
+ path = base_path.joinpath(filename)
log = logger.bind(
base_path=base_path,
filename=filename,
@@ -138,19 +137,19 @@ def _jenkins_get(url, base_path, filename):
def jenkins_get_console(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")
def jenkins_get_build(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")
def jenkins_get_env(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")
@@ -161,8 +160,8 @@ def jenkins_get_artifact(jobname, buildnumber, artifact_info):
buildnumber,
artifact_info["relativePath"],
)
- base_path = os.path.join(
- settings.JBI_BASEDIR, jobname, str(buildnumber), "artifact"
+ base_path = settings.JBI_BASEDIR.joinpath(
+ jobname, str(buildnumber), "artifact"
)
return _jenkins_get(url, base_path, artifact_info["fileName"])