Notice Publisher tests sync

Author: Kolea Plesco

dags/worker_single_notice_process_orchestrator.py

@@ -217,8 +217,8 @@ def _validate_transformed_rdf_manifestation(**context_args):
                                                notice_repository=notice_repository,
                                                mapping_suite_repository=mapping_suite_repository)
         validate_xpath_coverage_notice_by_id(notice_id=notice_id, mapping_suite_identifier=mapping_suite_id,
-                                             mapping_suite_repository=mapping_suite_repository,
-                                             mongodb_client=mongodb_client)
+                                             notice_repository=notice_repository,
+                                             mapping_suite_repository=mapping_suite_repository)
         push_dag_downstream(NOTICE_ID, notice_id)
         push_dag_downstream(MAPPING_SUITE_ID, mapping_suite_id)
 

infra/sftp/docker-compose.yml

@@ -8,9 +8,10 @@ services:
     environment:
       - SFTP_USER=${SFTP_USER}
       - SFTP_PASSWORD=${SFTP_PASSWORD}
+      - SFTP_PORT=${SFTP_PORT}
     command: ${SFTP_USER}:${SFTP_PASSWORD}:::upload
     ports:
-      - ${SFTP_PORT:-2235}:22
+      - ${SFTP_PORT:-2235}:${SFTP_PORT}
     networks:
       - sftp-net
       - proxy-net

ted_sws/__init__.py

@@ -156,8 +156,31 @@ def TED_API_URL(self) -> str:
         return EnvConfigResolver().config_resolve()
 
 
+class SFTPConfig:
+    @property
+    def SFTP_HOST(self) -> str:
+        return EnvConfigResolver().config_resolve()
+
+    @property
+    def SFTP_PORT(self) -> int:
+        v = EnvConfigResolver().config_resolve()
+        return int(v) if v is not None else 22
+
+    @property
+    def SFTP_USER(self) -> str:
+        return EnvConfigResolver().config_resolve()
+
+    @property
+    def SFTP_PASSWORD(self)->str:
+        return EnvConfigResolver().config_resolve()
+
+    @property
+    def SFTP_PATH(self)->str:
+        return EnvConfigResolver().config_resolve()
+
+
 class TedConfigResolver(MongoDBConfig, RMLMapperConfig, XMLProcessorConfig, ELKConfig, LoggingConfig,
-                        GitHubArtefacts, API, AllegroConfig, TedAPIConfig):
+                        GitHubArtefacts, API, AllegroConfig, TedAPIConfig, SFTPConfig):
     """
         This class resolve the secrets of the ted-sws project.
     """

ted_sws/notice_publisher/adapters/notice_publisher.py

@@ -1,7 +1,29 @@
-from abc import ABC
-from ted_sws.core.model.notice import Notice
+from enum import Enum
 
+from ted_sws.notice_publisher.adapters.notice_publisher_abc import NoticePublisherABC
+from ted_sws.notice_publisher.adapters.sftp_notice_publisher import SFTPNoticePublisher
+from ted_sws import config
 
-class NoticePublisherABC(ABC):
-    def upload(self, source_path, remote_path=None) -> bool:
-        pass
+
+class Publisher(Enum):
+    SFTP = "sftp"
+
+
+DEFAULT_PUBLISHER = Publisher.SFTP
+
+
+class NoticePublisherFactory:
+    @classmethod
+    def get_publisher(cls, publisher: Publisher = DEFAULT_PUBLISHER, **kwargs) -> NoticePublisherABC:
+        """Factory Method to return the needed Publisher, based on """
+
+        if publisher == Publisher.SFTP:
+            hostname = kwargs['hostname'] if 'hostname' in kwargs else config.SFTP_HOST
+            username = kwargs['username'] if 'username' in kwargs else config.SFTP_USER
+            password = kwargs['password'] if 'password' in kwargs else config.SFTP_PASSWORD
+            port = kwargs['port'] if 'port' in kwargs else config.SFTP_PORT
+            remote_path = kwargs['remote_path'] if 'remote_path' in kwargs else None
+            if not hostname or not username or not password:
+                raise Exception("SFTPPublisher mandatory init parameters: (host, username, password)")
+            return SFTPNoticePublisher(hostname=hostname, username=username, password=password, port=port,
+                                       remote_path=remote_path)

ted_sws/notice_publisher/adapters/notice_publisher_abc.py

@@ -0,0 +1,24 @@
+from abc import ABC
+
+
+class NoticePublisherABC(ABC):
+    def connect(self, **kwargs) -> bool:
+        """
+
+        :param kwargs:
+        :return:
+        """
+
+    def publish(self, **kwargs) -> bool:
+        """
+
+        :param kwargs:
+        :return:
+        """
+
+    def disconnect(self, **kwargs) -> bool:
+        """
+
+        :param kwargs:
+        :return:
+        """

ted_sws/notice_publisher/adapters/sftp_notice_publisher.py

@@ -1,19 +1,25 @@
 import pysftp
 
-from ted_sws.core.model.notice import Notice
-from ted_sws.notice_publisher.adapters.notice_publisher import NoticePublisherABC
+from ted_sws.notice_publisher.adapters.notice_publisher_abc import NoticePublisherABC
+from ted_sws import config
 
 
 class SFTPNoticePublisher(NoticePublisherABC):
-    def __init__(self, hostname, username, password, port=22, remote_path=None):
+    connection: pysftp.Connection = None
+    default_host = config.SFTP_HOST
+    default_user = config.SFTP_USER
+    default_pass = config.SFTP_PASSWORD
+    default_port = config.SFTP_PORT
+
+    def __init__(self, hostname=default_host, username=default_user, password=default_pass, port=default_port,
+                 remote_path=None):
         """Constructor Method"""
         # Set connection object to None (initial value)
-        self.connection = None
         self.hostname = hostname
         self.username = username
         self.password = password
+        self.port = port or self.default_port
         self.remote_path = remote_path
-        self.port = port
 
     def connect(self):
         """Connects to the sftp server and returns the sftp connection object"""
@@ -33,6 +39,9 @@ def disconnect(self):
         """Closes the sftp connection"""
         self.connection.close()
 
+    def publish(self, source_path, remote_path=None) -> bool:
+        return self.upload(source_path, remote_path)
+
     def upload(self, source_path, remote_path=None) -> bool:
         """
        Uploads the notice's METS manifestation to the sftp server remote path.
@@ -48,3 +57,10 @@ def upload(self, source_path, remote_path=None) -> bool:
             return True
         except Exception as err:
             raise Exception(err)
+
+    def remove(self, remote_path) -> bool:
+        try:
+            self.connection.unlink(remote_path)
+            return True
+        except Exception as err:
+            raise Exception(err)

ted_sws/notice_publisher/services/notice_publisher.py

@@ -1,34 +1,80 @@
-from ted_sws.notice_publisher.adapters.notice_publisher import NoticePublisherABC
-from ted_sws.notice_publisher.adapters.sftp_notice_publisher import SFTPNoticePublisher
-from ted_sws.core.model.notice import Notice, NoticeStatus
-from ted_sws.data_manager.adapters.notice_repository import NoticeRepositoryABC
 import base64
 import tempfile
+from pathlib import Path
+from typing import Union
+
+from ted_sws import config
+from ted_sws.core.model.notice import Notice, NoticeStatus, UnsupportedStatusTransition
+from ted_sws.data_manager.adapters.notice_repository import NoticeRepositoryABC
+from ted_sws.notice_packager.adapters.archiver import ARCHIVE_DEFAULT_FORMAT
+from ted_sws.notice_publisher.adapters.notice_publisher import NoticePublisherFactory
+from ted_sws.notice_publisher.adapters.notice_publisher_abc import NoticePublisherABC
+
+
+class NoticePublishBuilder:
+    def __init__(self, notice: Union[Notice, str], notice_repository: NoticeRepositoryABC,
+                 notice_publisher: NoticePublisherABC, remote_path: str = None):
+        self.notice_repository = notice_repository
+        if isinstance(notice, str):
+            notice_id = notice
+            notice: Notice = self.notice_repository.get(reference=notice_id)
+            if notice is None:
+                raise Exception(f"Notice {notice_id} could not be found.")
+
+        self.notice = notice
+        self.check_publish_eligibility()
 
+        self.notice_publisher = notice_publisher
+        self.remote_path = remote_path
 
-def publish_notice_package(notice: Notice, notice_publisher: NoticePublisherABC, notice_repository: NoticeRepositoryABC,
-                           remote_path=None):
-    mets_manifestation = notice.mets_manifestation
-    if not mets_manifestation:
-        raise ValueError("Notice does not have a METS manifestation to be published.")
+    def check_publish_eligibility(self):
+        if self.notice.status != NoticeStatus.ELIGIBLE_FOR_PUBLISHING:
+            raise UnsupportedStatusTransition(
+                f"Notice {self.notice.ted_id} is not Eligible for Publishing. Status: {self.notice.status}")
 
-    package_content = base64.b64decode(bytes(mets_manifestation.object_data), validate=True)
+    def package_content(self) -> bytes:
+        mets_manifestation = self.notice.mets_manifestation
+        if not mets_manifestation or not mets_manifestation.object_data:
+            raise ValueError("Notice does not have a METS manifestation to be published.")
 
-    source_file = tempfile.NamedTemporaryFile()
-    source_file.write(package_content)
+        package_content = base64.b64decode(bytes(mets_manifestation.object_data, encoding='utf-8'), validate=True)
+        return package_content
 
-    if notice_publisher.upload(source_file, remote_path):
-        notice.update_status_to(NoticeStatus.PUBLISHED)
-        notice_repository.update(notice)
-    else:
-        raise Exception(f"Notice {notice.ted_id} could not be published.")
+    def build_remote_path(self) -> str:
+        remote_path = self.remote_path
+        if remote_path is None:
+            remote_path = f"{config.SFTP_PATH}/{self.notice.ted_id}"
+            if ARCHIVE_DEFAULT_FORMAT == "zip":
+                remote_path += '.zip'
+        return remote_path
 
+    def publish(self):
+        source_file = tempfile.NamedTemporaryFile()
+        source_file.write(self.package_content())
 
-def publish_notice(notice: Notice, notice_repository: NoticeRepositoryABC, hostname, username, password,
-                   remote_path=None, port=22):
-    notice_publisher = SFTPNoticePublisher(hostname=hostname, username=username, password=password, port=port,
-                                           remote_path=remote_path)
+        try:
+            if self.notice_publisher.publish(source_path=Path(source_file.name), remote_path=self.build_remote_path()):
+                self.notice.update_status_to(NoticeStatus.PUBLISHED)
+                self.notice_repository.update(self.notice)
+        except Exception as e:
+            raise Exception(f"Notice {self.notice.ted_id} could not be published: " + str(e))
+
+        return self.notice.status == NoticeStatus.PUBLISHED
+
+
+def publish_notice(notice: Union[Notice, str], notice_publisher: NoticePublisherABC,
+                   notice_repository: NoticeRepositoryABC, remote_path=None) -> bool:
+    publish_builder = NoticePublishBuilder(notice, notice_repository, notice_publisher, remote_path)
+    return publish_builder.publish()
+
+
+def publish_single_notice(notice: Union[Notice, str], notice_repository: NoticeRepositoryABC, hostname, username,
+                          password, port=None, remote_path=None) -> bool:
+    notice_publisher: NoticePublisherABC = NoticePublisherFactory.get_publisher(hostname=hostname, username=username,
+                                                                                password=password, port=port,
+                                                                                remote_path=remote_path)
     notice_publisher.connect()
-    publish_notice_package(notice, notice_publisher, notice_repository)
+    notice_published: bool = publish_notice(notice, notice_publisher, notice_repository, remote_path)
     notice_publisher.disconnect()
-    pass
+
+    return notice_published

ted_sws/notice_validator/adapters/validation_summary_runner.py

@@ -1,7 +1,6 @@
 from typing import List
 
 from jinja2 import Environment, PackageLoader
-from pymongo import MongoClient
 
 from ted_sws.core.model.manifestation import ValidationSummaryReport, XMLManifestationValidationSummaryReport, \
     RDFManifestationValidationSummaryReport, XPATHCoverageSummaryReport, XPATHCoverageSummaryResult, \
@@ -123,8 +122,6 @@ class ValidationSummaryRunner:
         Runs Validation Summary
     """
 
-    mongodb_client: MongoClient
-
     def __init__(self):
         pass
 

ted_sws/notice_validator/services/validation_summary_runner.py

@@ -1,5 +1,3 @@
-from pymongo import MongoClient
-
 from ted_sws.core.model.manifestation import ValidationSummaryReport
 from ted_sws.core.model.notice import Notice
 from ted_sws.data_manager.adapters.notice_repository import NoticeRepository
@@ -29,9 +27,7 @@ def validation_summary_report_notice(notice: Notice):
     notice.validation_summary = report_builder.generate_report()
 
 
-def validation_summary_report_notice_by_id(notice_id: str,
-                                           mongodb_client: MongoClient):
-    notice_repository = NoticeRepository(mongodb_client=mongodb_client)
+def validation_summary_report_notice_by_id(notice_id: str, notice_repository: NoticeRepository):
     notice = notice_repository.get(reference=notice_id)
     if notice is None:
         raise ValueError(f'Notice, with {notice_id} id, was not found')

ted_sws/notice_validator/services/xpath_coverage_runner.py

@@ -1,7 +1,5 @@
 from typing import List
 
-from pymongo import MongoClient
-
 from ted_sws.core.model.manifestation import XPATHCoverageValidationReport
 from ted_sws.core.model.notice import Notice
 from ted_sws.core.model.transform import MappingSuite
@@ -27,10 +25,10 @@ def generate_report(self) -> XPATHCoverageValidationReport:
 def coverage_notice_xpath_report(notices: List[Notice], mapping_suite_id,
                                  conceptual_mappings_file_path: PATH_TYPE = None,
                                  coverage_runner: CoverageRunner = None, xslt_transformer=None,
-                                 mongodb_client: MongoClient = None) -> XPATHCoverageValidationReport:
+                                 notice_repository: NoticeRepository = None) -> XPATHCoverageValidationReport:
     if not coverage_runner:
         coverage_runner = CoverageRunner(mapping_suite_id, conceptual_mappings_file_path, xslt_transformer,
-                                         mongodb_client)
+                                         notice_repository.mongodb_client)
     report: XPATHCoverageValidationReport = coverage_runner.coverage_notice_xpath(notices, mapping_suite_id)
     return report
 
@@ -43,24 +41,23 @@ def xpath_coverage_html_report(report: XPATHCoverageValidationReport) -> str:
     return CoverageRunner.html_report(report)
 
 
-def validate_xpath_coverage_notice(notice: Notice, mapping_suite: MappingSuite, mongodb_client: MongoClient):
+def validate_xpath_coverage_notice(notice: Notice, mapping_suite: MappingSuite, notice_repository: NoticeRepository):
     xpath_coverage_report = coverage_notice_xpath_report(notices=[notice],
                                                          mapping_suite_id=mapping_suite.identifier,
-                                                         mongodb_client=mongodb_client)
+                                                         notice_repository=notice_repository)
     report_builder = XPATHCoverageReportBuilder(xpath_coverage_report=xpath_coverage_report)
     notice.set_xml_validation(xml_validation=report_builder.generate_report())
 
 
 def validate_xpath_coverage_notice_by_id(notice_id: str, mapping_suite_identifier: str,
                                          mapping_suite_repository: MappingSuiteRepositoryABC,
-                                         mongodb_client: MongoClient):
-    notice_repository = NoticeRepository(mongodb_client=mongodb_client)
+                                         notice_repository: NoticeRepository):
     notice = notice_repository.get(reference=notice_id)
     if notice is None:
         raise ValueError(f'Notice, with {notice_id} id, was not found')
 
     mapping_suite = mapping_suite_repository.get(reference=mapping_suite_identifier)
     if mapping_suite is None:
         raise ValueError(f'Mapping suite, with {mapping_suite_identifier} id, was not found')
-    validate_xpath_coverage_notice(notice=notice, mapping_suite=mapping_suite, mongodb_client=mongodb_client)
+    validate_xpath_coverage_notice(notice=notice, mapping_suite=mapping_suite, notice_repository=notice_repository)
     notice_repository.update(notice=notice)