test: add tests for celery app

.github/workflows/tests.yml

@@ -9,7 +9,7 @@ on:
 jobs:
 
   build:
-    if: ${{ github.event.workflow_run.conclusion == 'success' }}
+    # if: ${{ github.event.workflow_run.conclusion == 'success' }}
     runs-on: ubuntu-22.04
     environment: tests
     strategy:
@@ -84,13 +84,21 @@ jobs:
           sudo chown -R $(whoami):$(whoami) $VAR
           ./prepare_environment.sh
           sed -i "s/yc-redis/localhost/" $BACKEND/tests/resources/test.conf
+          sed -i "s/yc-rabbit/localhost/" $BACKEND/tests/resources/test.conf
+          sed -i "s/username=rabbitmq/username=guest/" $BACKEND/tests/resources/test.conf
+          sed -i "s/rabbitmq-password=rabbitmq/rabbitmq-password=guest/" $BACKEND/tests/resources/test.conf
 
       - name: Feed Redis
         run: |
           export BACKEND=$PWD
           export YANGCATALOG_CONFIG_PATH=$BACKEND/tests/resources/test.conf
           python feed_redis.py
 
+      - name: RabbitMQ in GitHub Actions
+        uses: Namoshek/[email protected]
+        with:
+          ports: "5672:5672"
+
       - name: Test with pytest
         env:
           TOKEN: ${{ secrets.TOKEN }}
@@ -100,6 +108,7 @@ jobs:
           export TEST_REPO=/var/yang/tmp/test/YangModels/yang
           export BACKEND=$PWD
           export YANGCATALOG_CONFIG_PATH=$BACKEND/tests/resources/test.conf
+          celery -A tests.test_celery_app:celery_app worker -l INFO &
           coverage run -am pytest
           coverage xml
 

jobs/app.py

@@ -0,0 +1,49 @@
+import logging
+import os
+
+import requests
+from celery import Celery
+
+from redisConnections.redisConnection import RedisConnection
+from utility import log
+from utility.create_config import create_config
+from utility.opensearch_util import ESIndexingPaths
+from utility.staticVariables import json_headers
+
+
+class BackendCeleryApp(Celery):
+    logger: logging.Logger
+    redis_connection: RedisConnection
+    notify_indexing: bool
+    save_file_dir: str
+    yangcatalog_api_prefix: str
+    indexing_paths: ESIndexingPaths
+    confd_credentials: tuple[str, str]
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.load_config()
+
+    def load_config(self):
+        config = create_config()
+        log_directory = config.get('Directory-Section', 'logs')
+        self.logger = log.get_logger('job_runner', os.path.join(log_directory, 'job_runner.log'))
+        self.redis_connection = RedisConnection(config=config)
+        self.notify_indexing = config.get('General-Section', 'notify-index') == 'True'
+        self.save_file_dir = config.get('Directory-Section', 'save-file-dir')
+        changes_cache_path = config.get('Directory-Section', 'changes-cache')
+        delete_cache_path = config.get('Directory-Section', 'delete-cache')
+        failed_changes_cache_path = config.get('Directory-Section', 'changes-cache-failed')
+        lock_file = config.get('Directory-Section', 'lock')
+        self.yangcatalog_api_prefix = config.get('Web-Section', 'yangcatalog-api-prefix')
+        self.indexing_paths = ESIndexingPaths(
+            cache_path=changes_cache_path,
+            deletes_path=delete_cache_path,
+            failed_path=failed_changes_cache_path,
+            lock_path=lock_file,
+        )
+        self.confd_credentials = tuple(config.get('Secrets-Section', 'confd-credentials').strip('"').split())
+        self.logger.info('Config loaded succesfully')
+
+    def reload_cache(self):
+        requests.post(f'{self.yangcatalog_api_prefix}/load-cache', auth=self.confd_credentials, headers=json_headers)

jobs/celery.py

@@ -19,59 +19,13 @@
 
 import functools
 import json
-import logging
-import os
 import shutil
 import typing as t
 
-import requests
-from celery import Celery
-
+from jobs.app import BackendCeleryApp
 from jobs.status_messages import StatusMessage
-from redisConnections.redisConnection import RedisConnection, key_quote
-from utility import log
-from utility.create_config import create_config
-from utility.opensearch_util import ESIndexingPaths, prepare_for_es_removal, send_for_es_indexing
-from utility.staticVariables import json_headers
-
-
-class BackendCeleryApp(Celery):
-    logger: logging.Logger
-    redis_connection: RedisConnection
-    notify_indexing: bool
-    save_file_dir: str
-    yangcatalog_api_prefix: str
-    indexing_paths: ESIndexingPaths
-    confd_credentials: tuple[str, str]
-
-    def __init__(self, *args, **kwargs):
-        super().__init__(*args, **kwargs)
-        self.load_config()
-
-    def load_config(self):
-        config = create_config()
-        log_directory = config.get('Directory-Section', 'logs')
-        self.logger = log.get_logger('job_runner', os.path.join(log_directory, 'job_runner.log'))
-        self.redis_connection = RedisConnection(config=config)
-        self.notify_indexing = config.get('General-Section', 'notify-index') == 'True'
-        self.save_file_dir = config.get('Directory-Section', 'save-file-dir')
-        changes_cache_path = config.get('Directory-Section', 'changes-cache')
-        delete_cache_path = config.get('Directory-Section', 'delete-cache')
-        failed_changes_cache_path = config.get('Directory-Section', 'changes-cache-failed')
-        lock_file = config.get('Directory-Section', 'lock')
-        self.yangcatalog_api_prefix = config.get('Web-Section', 'yangcatalog-api-prefix')
-        self.indexing_paths = ESIndexingPaths(
-            cache_path=changes_cache_path,
-            deletes_path=delete_cache_path,
-            failed_path=failed_changes_cache_path,
-            lock_path=lock_file,
-        )
-        self.confd_credentials = tuple(config.get('Secrets-Section', 'confd-credentials').strip('"').split())
-        self.logger.info('Config loaded succesfully')
-
-    def reload_cache(self):
-        requests.post(f'{self.yangcatalog_api_prefix}/load-cache', auth=self.confd_credentials, headers=json_headers)
-
+from redisConnections.redisConnection import key_quote
+from utility.opensearch_util import prepare_for_es_removal, send_for_es_indexing
 
 celery_app = BackendCeleryApp('backend_celery_app')
 

jobs/jobs_information.py

@@ -17,7 +17,7 @@
 __license__ = 'Apache License, Version 2.0'
 __email__ = '[email protected]'
 
-from jobs.celery import BackendCeleryApp
+from jobs.app import BackendCeleryApp
 from jobs.status_messages import StatusMessage, status_messages_mapping
 
 

tests/resources/test.conf

@@ -9,7 +9,7 @@ [email protected]
 
 [Secrets-Section]
 flask-secret-key=S3CR3T
-rabbitmq-password=guest
+rabbitmq-password=rabbitmq
 opensearch-secret=test
 confd-credentials='test test'
 yang-catalog-token=test
@@ -75,3 +75,9 @@ non-ietf-directory=/var/yang/nonietf
 email-from=test
 email-to=test
 developers-email=test
+
+[RabbitMQ-Section]
+host=yc-rabbit
+username=rabbitmq
+virtual-host=/
+port=5672

tests/test_celery_app.py

@@ -0,0 +1,55 @@
+# Copyright The IETF Trust 2023, All Rights Reserved
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+__author__ = 'Richard Zilincik'
+__copyright__ = 'Copyright The IETF Trust 2023, All Rights Reserved'
+__license__ = 'Apache License, Version 2.0'
+__email__ = '[email protected]'
+
+import os
+import time
+import unittest
+import uuid
+
+from jobs.app import BackendCeleryApp
+from jobs.jobs_information import get_response
+from jobs.status_messages import StatusMessage
+
+celery_app = BackendCeleryApp('celery_app')
+celery_app.config_from_object('jobs.celery_configuration')
+
+
+@celery_app.task
+def example_task(file: str):
+    open(file, 'w').close()
+    return
+
+
+class TestCeleryAppBaseClass(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.resources_path = os.path.join(os.environ['BACKEND'], 'tests/resources')
+
+    def test_success(self):
+        file = os.path.join(self.resources_path, str(uuid.uuid4()))
+        task_id = example_task.s(file=file).apply_async().id
+        time.sleep(1)
+        self.assertEqual(get_response(celery_app, task_id)[0], StatusMessage.SUCCESS.value)
+        self.assertTrue(os.path.exists(file))
+
+    def test_fail(self):
+        file = os.path.join('/', str(uuid.uuid4()))
+        task_id = example_task.s(file=file).apply_async().id
+        time.sleep(1)
+        self.assertEqual(get_response(celery_app, task_id)[0], StatusMessage.FAIL.value)