From 6f4bf0cc410b1785b4740c31457f13999efb3d26 Mon Sep 17 00:00:00 2001
From: Markus Scheidgen <markus.scheidgen@gmail.com>
Date: Sun, 12 Aug 2018 14:45:39 +0200
Subject: [PATCH] Migrated tests to pytest. Added typesafety.
---
.gitignore | 3 +-
requirements-dev.txt | 4 +-
tests/test_dependencies.py | 16 ++-----
tests/test_files.py | 91 +++++++++++++++++++-------------------
tests/test_processing.py | 46 ++++++++-----------
5 files changed, 72 insertions(+), 88 deletions(-)
diff --git a/.gitignore b/.gitignore
index 22ca8ca382..fd472b4757 100644
--- a/.gitignore
+++ b/.gitignore
@@ -5,4 +5,5 @@ __pycache__
*.egg-info/
data
.dependencies/
-.volumes/
\ No newline at end of file
+.volumes/
+.pytest_cache/
\ No newline at end of file
diff --git a/requirements-dev.txt b/requirements-dev.txt
index 74182b2ac4..27caaf5c7b 100644
--- a/requirements-dev.txt
+++ b/requirements-dev.txt
@@ -4,4 +4,6 @@ recommonmark
gitpython
mypy
pylint
-pycodestyle
\ No newline at end of file
+pycodestyle
+pytest
+pytest-timeout
\ No newline at end of file
diff --git a/tests/test_dependencies.py b/tests/test_dependencies.py
index 55bfb635e4..d32f5ec038 100644
--- a/tests/test_dependencies.py
+++ b/tests/test_dependencies.py
@@ -1,16 +1,6 @@
-import unittest
-from unittest import TestCase
-import logging
-
from nomad.dependencies import parser_dict
-class DependenciesTests(TestCase):
-
- def test_vasp_parser(self):
- vasp_parser = parser_dict['parsers/vasp']
- vasp_parser.run('.dependencies/parsers/vasp/test/examples/xml/perovskite.xml')
-
-if __name__ == '__main__':
- logging.basicConfig(level=logging.INFO)
- unittest.main()
+def test_vasp_parser():
+ vasp_parser = parser_dict['parsers/vasp']
+ vasp_parser.run('.dependencies/parsers/vasp/test/examples/xml/perovskite.xml')
diff --git a/tests/test_files.py b/tests/test_files.py
index 1e0a7bea70..2ebb73950f 100644
--- a/tests/test_files.py
+++ b/tests/test_files.py
@@ -1,72 +1,71 @@
-import unittest
+from typing import Generator
+import pytest
import time
from minio import ResponseError
-import logging
-from unittest import TestCase
from threading import Thread
import subprocess
import shlex
-import logging
import nomad.files as files
import nomad.config as config
-LOGGER = logging.getLogger(__name__)
-test_upload_id = '__test_upload_id'
+example_file = './data/examples_vasp.zip'
+example_upload_id = '__test_upload_id'
-def upload_test_file():
- example_file = './data/examples_vasp.zip'
- upload_url = files.get_presigned_upload_url(test_upload_id)
- cmd = files.create_curl_upload_cmd(upload_url).replace('<ZIPFILE>', example_file)
- LOGGER.debug('Initiate upload of example file %s with command %s' % (example_file, cmd))
- subprocess.call(shlex.split(cmd))
+@pytest.fixture
+def uploaded_id() -> str:
+ files._client.fput_object(config.s3.uploads_bucket, example_upload_id, example_file)
+ return example_upload_id
-class FilesTests(TestCase):
+@pytest.fixture(autouse=True)
+def tear_down():
+ yield
+ try:
+ files._client.remove_object(config.s3.uploads_bucket, example_upload_id)
+ except ResponseError:
+ pass
- def tearDown(self):
- try:
- files._client.remove_object(config.s3.uploads_bucket, test_upload_id)
- except ResponseError:
- pass
- def test_presigned_url(self):
- url = files.get_presigned_upload_url(test_upload_id)
+def test_presigned_url():
+ url = files.get_presigned_upload_url(example_upload_id)
+ assert url is not None
+ assert isinstance(url, str)
- self.assertIsNotNone(url)
- self.assertIsInstance(url, str)
+ upload_url = files.get_presigned_upload_url(example_upload_id)
+ cmd = files.create_curl_upload_cmd(upload_url).replace('<ZIPFILE>', example_file)
+ subprocess.call(shlex.split(cmd))
- def test_upload(self):
- upload_test_file()
+ stat = files._client.stat_object(config.s3.uploads_bucket, example_upload_id)
+ assert stat.content_type.startswith('application/octet-steam')
- with files.Upload(test_upload_id) as upload:
- self.assertEqual(106, len(upload.filelist))
- # now just try to open the first file (not directory), without error
- for filename in upload.filelist:
- if filename.endswith('.xml'):
- upload.open_file(filename).close()
- break
- def test_upload_notification(self):
- @files.upload_put_handler
- def handle_upload_put(upload_id):
- self.assertEqual(test_upload_id, upload_id)
- raise StopIteration
+def test_upload(uploaded_id: str):
+ with files.Upload(uploaded_id) as upload:
+ assert len(upload.filelist) == 106
+ # now just try to open the first file (not directory), without error
+ for filename in upload.filelist:
+ if filename.endswith('.xml'):
+ upload.open_file(filename).close()
+ break
- def handle_uploads():
- handle_upload_put(upload_id='provided by decorator')
- handle_uploads_thread = Thread(target=handle_uploads)
- handle_uploads_thread.start()
+@pytest.mark.timeout(10)
+def test_upload_notification():
+ @files.upload_put_handler
+ def handle_upload_put(upload_id: str):
+ assert upload_id == example_upload_id
+ raise StopIteration
- time.sleep(1)
- upload_test_file()
+ def handle_uploads():
+ handle_upload_put(upload_id='provided by decorator')
- handle_uploads_thread.join()
+ handle_uploads_thread = Thread(target=handle_uploads)
+ handle_uploads_thread.start()
+ time.sleep(1)
+ test_presigned_url()
-if __name__ == '__main__':
- logging.basicConfig(level=logging.DEBUG)
- unittest.main()
+ handle_uploads_thread.join()
diff --git a/tests/test_processing.py b/tests/test_processing.py
index 46d67d717b..a2b6782402 100644
--- a/tests/test_processing.py
+++ b/tests/test_processing.py
@@ -1,42 +1,34 @@
-import unittest
-from unittest import TestCase
-import logging
+from typing import Generator
+import pytest
from minio import ResponseError
-import os
import nomad.files as files
import nomad.config as config
from nomad.processing import UploadProcessing
-test_upload_id = '__test_upload_id'
+example_upload_id = '__test_upload_id'
+example_file = 'data/examples_vasp.zip'
-class ProcessingTests(TestCase):
+@pytest.fixture
+def uploaded_id() -> Generator[str, None, None]:
+ files._client.fput_object(config.s3.uploads_bucket, example_upload_id, example_file)
- def setUp(self):
- example_file = 'data/examples_vasp.zip'
- with open(example_file, 'rb') as file_data:
- file_stat = os.stat(example_file)
- files._client.put_object(
- config.s3.uploads_bucket, test_upload_id, file_data, file_stat.st_size)
+ yield example_upload_id
- def tearDown(self):
- try:
- files._client.remove_object(config.s3.uploads_bucket, test_upload_id)
- except ResponseError:
- pass
+ try:
+ files._client.remove_object(config.s3.uploads_bucket, example_upload_id)
+ except ResponseError:
+ pass
- def test_processing(self):
- run = UploadProcessing(test_upload_id)
+
+def test_processing(uploaded_id):
+ run = UploadProcessing(uploaded_id)
run.start()
run.get(timeout=30)
- self.assertTrue(run.ready())
- run.forget()
-
- self.assertEqual('nomad.processing.close_upload', run.task_name)
- self.assertEqual('SUCCESS', run.status)
+ assert run.ready()
+ run.forget()
-if __name__ == '__main__':
- logging.basicConfig(level=logging.INFO)
- unittest.main()
+ assert run.task_name == 'nomad.processing.close_upload'
+ assert run.status == 'SUCCESS'
--
GitLab