some tests

Author: amyasnikov

validity/api/helpers.py

@@ -60,7 +60,7 @@ def model_perms(*permissions: str) -> type[BasePermission]:
 
     class Permission(BasePermission):
         def has_permission(self, request, view):
-            return request.user.is_authenticated and request.user.has_perms([permissions])
+            return request.user.is_authenticated and request.user.has_perms(permissions)
 
     return Permission
 

validity/data_backup/backupers.py

@@ -39,7 +39,7 @@ class GitBackuper(Backuper):
 
     def _get_repo(self, url: str, parameters: GitParams, datasource_dir: Path) -> RemoteGitRepo:
         return RemoteGitRepo(
-            local_path=datasource_dir,
+            local_path=str(datasource_dir),
             remote_url=url,
             active_branch=parameters.branch,
             username=parameters.username,

validity/integrations/errors.py

@@ -15,4 +15,4 @@ def __str__(self):
     def __repr__(self):
         if self.orig_error:
             return repr(self.orig_error)
-        return super().__repr__(self)
+        return super().__repr__()

validity/integrations/git.py

@@ -25,13 +25,13 @@ def clone(
     ) -> None: ...
 
     @abstractmethod
-    def stage_all(self, repo_path: str) -> None: ...
+    def stage_all(self, local_path: str) -> None: ...
 
     @abstractmethod
     def unstage_all(self, local_path: str) -> None: ...
 
     @abstractmethod
-    def commit(self, repo_path: str, username: str, email: str, message: str) -> str:
+    def commit(self, local_path: str, username: str, email: str, message: str) -> str:
         """
         Returns SHA1 hash of the new commit
         """
@@ -67,11 +67,11 @@ def clone(
         with reraise(Exception, IntegrationError):
             porcelain.clone(remote_url, local_path, checkout=checkout, **optional_args)
 
-    def stage_all(self, repo_path: str) -> None:
-        repo = Repo(repo_path)
+    def stage_all(self, local_path: str) -> None:
+        repo = Repo(local_path)
         ignore_mgr = IgnoreFilterManager.from_repo(repo)
-        unstaged_files = (fn.decode() for fn in get_unstaged_changes(repo.open_index(), repo_path))
-        untracked_files = porcelain.get_untracked_paths(repo_path, repo_path, repo.open_index())
+        unstaged_files = (fn.decode() for fn in get_unstaged_changes(repo.open_index(), local_path))
+        untracked_files = porcelain.get_untracked_paths(local_path, local_path, repo.open_index())
         files = (file for file in chain(unstaged_files, untracked_files) if not ignore_mgr.is_ignored(file))
         repo.stage(files)
 
@@ -80,9 +80,9 @@ def unstage_all(self, local_path: str) -> None:
         staged_files = chain.from_iterable(porcelain.status(local_path).staged.values())
         repo.unstage(filename.decode() for filename in staged_files)
 
-    def commit(self, repo_path: str, username: str, email: str, message: str) -> str:
+    def commit(self, local_path: str, username: str, email: str, message: str) -> str:
         author = f"{username} <{email}>".encode()
-        commit_hash = porcelain.commit(repo=repo_path, author=author, committer=author, message=message.encode())
+        commit_hash = porcelain.commit(repo=local_path, author=author, committer=author, message=message.encode())
         return commit_hash.decode()
 
     def push(

validity/models/backup.py

@@ -11,7 +11,7 @@
 from validity import di
 from validity.choices import BackupMethodChoices, BackupStatusChoices
 from validity.data_backup import BackupBackend
-from validity.fields import EncryptedDict, EncryptedDictField
+from validity.fields import EncryptedDictField
 from validity.subforms import GitBackupForm, S3BackupForm
 from .base import BaseModel, SubformMixin
 from .data import VDataSource
@@ -79,9 +79,7 @@ def get_last_status_color(self):
         return BackupStatusChoices.colors.get(self.last_status)
 
     def serialize_object(self, exclude=None):
-        if not isinstance(self.parameters, EncryptedDict):
-            do_not_encrypt = self._meta.get_field("parameters").do_not_encrypt
-            self.parameters = EncryptedDict(self.parameters, do_not_encrypt=do_not_encrypt)
+        self.parameters = self._meta.get_field("parameters").to_python(self.parameters)
         return super().serialize_object(exclude)
 
     def do_backup(self) -> None:

validity/scripts/keeper.py

@@ -25,13 +25,13 @@ def __enter__(self):
         return self
 
     def __exit__(self, exc_type, exc, tb):
-        if exc_type:
-            self.terminate_errored_job(exc)
-        elif self.job.status == JobStatusChoices.STATUS_RUNNING and self.auto_terminate:
-            self.terminate_job()
-        self.logger.flush()
+        with self.logger:
+            if exc_type:
+                return self.terminate_errored_job(exc)
+            elif self.job.status == JobStatusChoices.STATUS_RUNNING and self.auto_terminate:
+                self.terminate_job()
 
-    def terminate_errored_job(self, error: Exception) -> None:
+    def terminate_errored_job(self, error: Exception) -> bool:
         if isinstance(error, AbortScript):
             self.logger.messages.extend(error.logs)
             self.logger.failure(str(error))
@@ -41,6 +41,7 @@ def terminate_errored_job(self, error: Exception) -> None:
             status = JobStatusChoices.STATUS_ERRORED
         self.error_callback(self, error)
         self.terminate_job(status=status, error=repr(error))
+        return isinstance(error, AbortScript)
 
     def terminate_job(
         self, status: str = JobStatusChoices.STATUS_COMPLETED, error: str | None = None, output=None

validity/tests/conftest.py

@@ -4,6 +4,7 @@
 from core.models import DataSource
 from dcim.models import Device, DeviceType, Manufacturer
 from django.contrib.contenttypes.models import ContentType
+from django.utils import timezone
 from extras.models import CustomField
 from graphene_django.utils.testing import graphql_query
 from tenancy.models import Tenant
@@ -88,3 +89,11 @@ def func(*args, **kwargs):
 @pytest.fixture
 def di():
     return validity.di
+
+
+@pytest.fixture
+def timezone_now(monkeypatch):
+    def _now(tz):
+        monkeypatch.setattr(timezone, "now", lambda: tz)
+
+    return _now

validity/tests/factories.py

@@ -13,6 +13,7 @@
 
 from validity import models
 from validity.compliance.state import StateItem
+from validity.fields.encrypted import EncryptedDict
 
 
 DJANGO_MAJOR_VERSION = django.VERSION[:2]
@@ -28,6 +29,10 @@ class Meta:
         model = models.VDataSource
 
 
+class PollingDSFactory(DataSourceFactory):
+    type = "device_polling"
+
+
 class DataFileFactory(DjangoModelFactory):
     source = factory.SubFactory(DataSourceFactory)
     path = factory.Sequence(lambda n: f"file-{n}.txt")
@@ -236,6 +241,18 @@ class Meta:
         model = models.Poller
 
 
+class BackupPointFactory(DjangoModelFactory):
+    name = factory.sequence(lambda n: f"bp-{n}")
+    data_source = factory.SubFactory(PollingDSFactory)
+    backup_after_sync = False
+    method = "git"
+    url = "https://site.com/project"
+    parameters = EncryptedDict({"username": "a", "password": "b"})
+
+    class Meta:
+        model = models.BackupPoint
+
+
 class UserFactory(DjangoModelFactory):
     email = "[email protected]"
     username = "su"
@@ -261,6 +278,18 @@ class Meta:
         model = Job
 
 
+class DSBackupJobFactory(DjangoModelFactory):
+    name = "DataSourcebackup"
+    object = factory.SubFactory(BackupPointFactory)
+    object_id = factory.SelfAttribute("object.pk")
+    object_type = factory.LazyAttribute(lambda obj: ContentType.objects.get_for_model(type(obj.object)))
+    user = factory.SubFactory(UserFactory)
+    job_id = factory.LazyFunction(uuid.uuid4)
+
+    class Meta:
+        model = Job
+
+
 _NOT_DEFINED = object()
 
 

validity/tests/test_api.py

@@ -4,16 +4,19 @@
 import pytest
 from base import ApiGetTest, ApiPostGetTest
 from factories import (
+    BackupPointFactory,
     CommandFactory,
     CompTestDBFactory,
     CompTestResultFactory,
     DataFileFactory,
     DataSourceFactory,
     DeviceFactory,
     DeviceTypeFactory,
+    DSBackupJobFactory,
     LocationFactory,
     ManufacturerFactory,
     PlatformFactory,
+    PollingDSFactory,
     ReportFactory,
     RunTestsJobFactory,
     SelectorFactory,
@@ -26,6 +29,7 @@
 
 from validity import dependencies
 from validity.models import VDevice
+from validity.scripts import Launcher
 
 
 class TestDBNameSet(ApiPostGetTest):
@@ -173,6 +177,29 @@ class TestPoller(ApiPostGetTest):
     }
 
 
+class TestBackupPoint(ApiPostGetTest):
+    entity = "backup-points"
+    post_body = {
+        "name": "bp",
+        "data_source": PollingDSFactory,
+        "backup_after_sync": False,
+        "method": "git",
+        "upload_url": "http://ex.com/qwer",
+        "parameters": {"username": "abc", "password": "123"},
+    }
+
+    @pytest.mark.django_db
+    def test_backup(self, admin_client, di):
+        bp = BackupPointFactory()
+        url = self.url(bp.pk) + "backup/"
+        launcher = Mock(spec=Launcher, return_value=DSBackupJobFactory())
+        with di.override({dependencies.backup_launcher: lambda: launcher}):
+            resp = admin_client.post(url)
+            assert resp.status_code == HTTPStatus.OK
+            assert resp.json()["result"]["id"] == launcher.return_value.pk
+            launcher.assert_called_once()
+
+
 @pytest.mark.parametrize("params", [{}, {"fields": ["name", "value"]}, {"name": ["config", "bad_cmd"]}])
 @pytest.mark.django_db
 def test_get_serialized_state(admin_client, params, monkeypatch):