wip

requirements/base.txt

@@ -1,7 +1,6 @@
 django-bootstrap-v5==1.0.*
 pydantic==1.10.*
 ttp==0.9.*
-pygit2==1.11.*
 jq==1.4.*
 deepdiff==6.2.*
 simpleeval==0.9.*

validity/config_compliance/device_config/base.py

@@ -1,53 +1,40 @@
 from abc import abstractmethod
 from dataclasses import dataclass
 from datetime import datetime
-from pathlib import Path
 from typing import ClassVar
 
 from dcim.models import Device
-from django.utils.timezone import make_aware
 
-from validity import settings
 from validity.utils.misc import reraise
 from ..exceptions import DeviceConfigError
 
 
 @dataclass
 class BaseDeviceConfig:
     device: Device
-    config_path: Path
+    plain_config: str
     last_modified: datetime | None = None
     serialized: dict | list | None = None
-    _git_folder: ClassVar[Path] = settings.git_folder
-    _config_classes: ClassVar[dict[str, type]] = {}
 
-    @classmethod
-    def _full_config_path(cls, device: Device) -> Path:
-        return cls._git_folder / device.repo.name / device.repo.rendered_device_path(device)
+    _config_classes: ClassVar[dict[str, type]] = {}
 
     @classmethod
     def from_device(cls, device: Device) -> "BaseDeviceConfig":
         """
         Get DeviceConfig from dcim.models.Device
-        Device MUST be annotated with ".repo" pointing to a repo with device config file
+        Device MUST be annotated with ".plain_config"
         Device MUST be annotated with ".serializer" pointing to appropriate config serializer instance
         """
-        with reraise((AssertionError, FileNotFoundError), DeviceConfigError):
-            assert getattr(device, "repo", None), f"{device} has no bound repository"
+        with reraise((AssertionError, FileNotFoundError, AttributeError), DeviceConfigError):
+            assert getattr(device, "data_file", None), f"{device} has no bound data file"
             assert getattr(device, "serializer", None), f"{device} has no bound serializer"
             return cls._config_classes[device.serializer.extraction_method]._from_device(device)
 
     @classmethod
-    def _from_device(cls, device: Device) -> "BaseDeviceConfig":
-        with reraise(AttributeError, DeviceConfigError):
-            device_path = cls._full_config_path(device)
-            last_modified = None
-            if device_path.is_file():
-                lm_timestamp = device_path.stat().st_mtime
-                last_modified = make_aware(datetime.fromtimestamp(lm_timestamp))
-            instance = cls(device, device_path, last_modified)
-            instance.serialize()
-            return instance
+    def _from_device(cls, device: Device) -> "BaseDeviceConfig":         
+        instance = cls(device, device.data_file.data_as_string, device.data_file.last_updated)
+        instance.serialize()
+        return instance
 
     @abstractmethod
     def serialize(self, override: bool = False) -> None:

validity/config_compliance/device_config/routeros.py

@@ -1,7 +1,7 @@
 import logging
 import re
 from dataclasses import dataclass, field
-from pathlib import Path
+import io
 from typing import ClassVar, Generator, Literal
 
 from validity.utils.misc import reraise
@@ -105,45 +105,45 @@ def from_plain_text(cls, line: str) -> "ParsedLine":
         return cls(method=method, find_by=find, properties=properties, implicit_name=implicit_name)
 
 
-def parse_config(filename: str | Path) -> dict:
+def parse_config(plain_config: str) -> dict:
     result = {}
     context_path = []
     prevlines = []
-    with open(filename, "r") as cfgfile:
-        for line_num, line in enumerate(cfgfile, start=1):
-            if line.startswith(("#", ":")) or line == "\n":
-                continue
-            if line.startswith("/"):
-                context_path = line[1:-1].split()
-                continue
-            if line.endswith("\\\n"):
-                prevlines.append(line)
-                continue
-            if prevlines:
-                line = "".join(prevlines) + line
-                prevlines = []
+    cfgfile = io.StringIO(plain_config)
+    for line_num, line in enumerate(cfgfile, start=1):
+        if line.startswith(("#", ":")) or line == "\n":
+            continue
+        if line.startswith("/"):
+            context_path = line[1:-1].split()
+            continue
+        if line.endswith("\\\n"):
+            prevlines.append(line)
+            continue
+        if prevlines:
+            line = "".join(prevlines) + line
+            prevlines = []
+        try:
+            parsed_line = ParsedLine.from_plain_text(line)
+        except LineParsingError as e:
+            e.args = (e.args[0] + f", config line {line_num}",) + e.args[1:]
+            raise
+        current_context = result
+        for key in context_path:
             try:
-                parsed_line = ParsedLine.from_plain_text(line)
-            except LineParsingError as e:
-                e.args = (e.args[0] + f", config line {line_num}",) + e.args[1:]
-                raise
-            current_context = result
-            for key in context_path:
-                try:
-                    current_context = current_context[key]
-                except KeyError:
-                    current_context[key] = {}
-                    current_context = current_context[key]
-            if parsed_line.find_by or parsed_line.method == "add" or parsed_line.implicit_name:
-                if "values" not in current_context:
-                    current_context["values"] = []
-                current_context["values"].append(parsed_line.properties)
-                if parsed_line.find_by:
-                    current_context["values"][-1]["find_by"] = [
-                        {"key": parsed_line.find_by[0], "value": parsed_line.find_by[1]}
-                    ]
-            else:
-                current_context["properties"] = parsed_line.properties
+                current_context = current_context[key]
+            except KeyError:
+                current_context[key] = {}
+                current_context = current_context[key]
+        if parsed_line.find_by or parsed_line.method == "add" or parsed_line.implicit_name:
+            if "values" not in current_context:
+                current_context["values"] = []
+            current_context["values"].append(parsed_line.properties)
+            if parsed_line.find_by:
+                current_context["values"][-1]["find_by"] = [
+                    {"key": parsed_line.find_by[0], "value": parsed_line.find_by[1]}
+                ]
+        else:
+            current_context["properties"] = parsed_line.properties
     return result
 
 
@@ -153,4 +153,4 @@ class RouterOSDeviceConfig(DeviceConfig):
     def serialize(self, override: bool = False) -> None:
         if not self.serialized or override:
             with reraise(Exception, DeviceConfigError):
-                self.serialized = parse_config(self.config_path)
+                self.serialized = parse_config(self.plain_config)

validity/config_compliance/device_config/ttp.py

@@ -26,9 +26,7 @@ def __post_init__(self):
 
     def serialize(self, override: bool = False) -> None:
         if not self.serialized or override:
-            if not self.config_path.is_file():
-                raise DeviceConfigError(f"{self.config_path} does not exist")
-            parser = ttp(data=str(self.config_path), template=self._template.template)
+            parser = ttp(data=self.plain_config, template=self._template.template)
             parser.parse()
             with reraise(
                 IndexError, DeviceConfigError, msg=f"Invalid parsed config for {self.device}: {parser.result()}"

validity/config_compliance/device_config/yaml.py

@@ -12,10 +12,9 @@ class YAMLDeviceConfig(DeviceConfig):
 
     def serialize(self, override: bool = False) -> None:
         if not self.serialized or override:
-            with self.config_path.open("r") as cfg_file:
-                with reraise(
-                    yaml.YAMLError,
-                    DeviceConfigError,
-                    msg=f"Trying to parse invalid YAML as device config for {self.device}",
-                ):
-                    self.serialized = yaml.safe_load(cfg_file)
+            with reraise(
+                yaml.YAMLError,
+                DeviceConfigError,
+                msg=f"Trying to parse invalid YAML as device config for {self.device}",
+            ):
+                self.serialized = yaml.safe_load(self.plain_config)

validity/managers.py

@@ -29,15 +29,6 @@
     from validity.models.base import BaseModel
 
 
-class JSONObjMixin:
-    def as_json(self):
-        return self.values(json=JSONObject(**{f: f for f in self.model.json_fields}))
-
-
-class GitRepoQS(JSONObjMixin, RestrictedQuerySet):
-    pass
-
-
 class ComplianceTestQS(RestrictedQuerySet):
     def pf_latest_results(self) -> "ComplianceTestQS":
         from validity.models import ComplianceTestResult
@@ -80,6 +71,11 @@ def delete_old(self):
         return self.filter(report=None).last_more_than(settings.store_last_results).delete()
 
 
+class JSONObjMixin:
+    def as_json(self):
+        return self.values(json=JSONObject(**{f: f for f in self.model.json_fields}))
+
+
 class ConfigSerializerQS(JSONObjMixin, RestrictedQuerySet):
     pass
 
@@ -96,6 +92,10 @@ def percentage(field1: str, field2: str) -> Case:
     )
 
 
+class VDataFileQS(JSONObjMixin, RestrictedQuerySet):
+    pass
+
+
 class ComplianceReportQS(RestrictedQuerySet):
     def annotate_result_stats(self, groupby_field: DeviceGroupByChoices | None = None):
         qs = self
@@ -158,15 +158,15 @@ def _fetch_all(self):
                 if isinstance(item, self.model):
                     item.selector = self.selector
 
-    def annotate_git_repo_id(self: _QS) -> _QS:
-        from validity.models import GitRepo
+    def annotate_datasource_id(self: _QS) -> _QS:
+        from validity.models import VDataSource
 
         return self.annotate(
-            bound_repo=Cast(KeyTextTransform("repo", "tenant__custom_field_data"), BigIntegerField())
+            bound_source=Cast(KeyTextTransform("config_data_source", "tenant__custom_field_data"), BigIntegerField())
         ).annotate(
-            repo_id=Case(
-                When(bound_repo__isnull=False, then=F("bound_repo")),
-                default=GitRepo.objects.filter(default=True).values("id")[:1],
+            datasource_id=Case(
+                When(bound_source__isnull=False, then=F("bound_source")),
+                default=VDataSource.objects.filter(custom_field_data__device_config_default=True).values("id")[:1],
                 output_field=BigIntegerField(),
             )
         )
@@ -187,19 +187,13 @@ def annotate_serializer_id(self: _QS) -> _QS:
             )
         )
 
-    def annotate_json_serializer_repo(self: _QS) -> _QS:
-        from validity.models import GitRepo
+    def annotate_serializer_data_file(self: _QS) -> _QS:
+        from validity.models import VDataFile
 
         return self.annotate(
-            json_serializer_repo=GitRepo.objects.filter(configserializer__pk=OuterRef("serializer_id")).as_json()
+            serializer_data=VDataFile.objects.filter(configserializer__pk=OuterRef("serializer_id")).values('data')[:1]
         )
 
-    def annotate_json_repo(self: _QS) -> _QS:
-        from validity.models import GitRepo
-
-        qs = self.annotate_git_repo_id()
-        return annotate_json(qs, "repo", GitRepo)
-
     def annotate_json_serializer(self: _QS) -> _QS:
         from validity.models import ConfigSerializer
 
@@ -213,9 +207,6 @@ def _count_per_something(self, field: str, annotate_method: str) -> dict[int | N
             result[values[field]] = values["cnt"]
         return result
 
-    def count_per_repo(self) -> dict[int | None, int]:
-        return self._count_per_something("repo_id", "annotate_git_repo_id")
-
     def count_per_serializer(self) -> dict[int | None, int]:
         return self._count_per_something("serializer_id", "annotate_serializer_id")