yujunjun/OpenMetadata
1
0
Fork 0
mirror of https://github.com/open-metadata/OpenMetadata.git synced 2025-11-03 03:59:12 +00:00
Code Issues Packages Projects Releases Wiki Activity

Issue - 1091 (#1096)

Browse Source 
* fixed issue with not being able to load config file; cleaned up unused code

* fixed test case to test configuration error

Co-authored-by: Vijay Mariadassou <vijay@mariadassou.com>
This commit is contained in:
vijaypm 2021-11-07 11:19:06 -08:00 committed by GitHub
parent 3489cbbfb1
commit d14755b4e1
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 7 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
ingestion/tests/unit/workflow_test.py
View File

@ -2,7 +2,7 @@ import importlib
import pathlib import pathlib
from unittest import TestCase from unittest import TestCase
from metadata.config.common import load_config_file from metadata.config.common import load_config_file, ConfigurationError
from metadata.ingestion.api.workflow import Workflow from metadata.ingestion.api.workflow import Workflow
from metadata.ingestion.ometa.ometa_api import OpenMetadata from metadata.ingestion.ometa.ometa_api import OpenMetadata
from metadata.ingestion.ometa.openmetadata_rest import MetadataServerConfig from metadata.ingestion.ometa.openmetadata_rest import MetadataServerConfig
@ -49,7 +49,8 @@ class WorkflowTest(TestCase):
self.assertEqual(replace, "QueryParser") self.assertEqual(replace, "QueryParser")
def test_execute_200(self): def test_execute_200(self):
config_file = pathlib.Path("tests/unit/mysql_test.json") current_dir = pathlib.Path(__file__).resolve().parent
config_file = current_dir.joinpath("mysql_test.json")
workflow_config = load_config_file(config_file) workflow_config = load_config_file(config_file)
workflow = Workflow.create(workflow_config) workflow = Workflow.create(workflow_config)
workflow.execute() workflow.execute()
@ -69,12 +70,8 @@ class WorkflowTest(TestCase):
self.assertEqual(ingestionData is not None, True) self.assertEqual(ingestionData is not None, True)
def test_execute_4xx(self): def test_execute_4xx(self):
config_file = pathlib.Path("tests/unit/mysql_test.json") config_file = pathlib.Path("/tmp/mysql_test123")
workflow_config = load_config_file(config_file)
ingestionData = None
try: try:
file_path = "/tmp/mysql_test123" workflow_config = load_config_file(config_file)
with open(file_path) as ingestionFile: except ConfigurationError:
ingestionData = ingestionFile.read() self.assertRaises(ConfigurationError)
except FileNotFoundError:
self.assertRaises(FileNotFoundError)