test_database.py 4.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127
  1. # Licensed to the Apache Software Foundation (ASF) under one
  2. # or more contributor license agreements. See the NOTICE file
  3. # distributed with this work for additional information
  4. # regarding copyright ownership. The ASF licenses this file
  5. # to you under the Apache License, Version 2.0 (the
  6. # "License"); you may not use this file except in compliance
  7. # with the License. You may obtain a copy of the License at
  8. #
  9. # http://www.apache.org/licenses/LICENSE-2.0
  10. #
  11. # Unless required by applicable law or agreed to in writing,
  12. # software distributed under the License is distributed on an
  13. # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  14. # KIND, either express or implied. See the License for the
  15. # specific language governing permissions and limitations
  16. # under the License.
  17. """Test Task Database."""
  18. from unittest.mock import patch
  19. import pytest
  20. from pydolphinscheduler.tasks.database import Database
  21. TEST_DATABASE_TASK_TYPE = "SQL"
  22. TEST_DATABASE_SQL = "select 1"
  23. TEST_DATABASE_DATASOURCE_NAME = "test_datasource"
  24. @patch(
  25. "pydolphinscheduler.core.task.Task.gen_code_and_version",
  26. return_value=(123, 1),
  27. )
  28. @patch(
  29. "pydolphinscheduler.tasks.database.Database.get_datasource_info",
  30. return_value=({"id": 1, "type": "mock_type"}),
  31. )
  32. def test_get_datasource_detail(mock_datasource, mock_code_version):
  33. """Test :func:`get_datasource_type` and :func:`get_datasource_id` can return expect value."""
  34. name = "test_get_database_detail"
  35. task = Database(
  36. TEST_DATABASE_TASK_TYPE, name, TEST_DATABASE_DATASOURCE_NAME, TEST_DATABASE_SQL
  37. )
  38. assert 1 == task.get_datasource_id()
  39. assert "mock_type" == task.get_datasource_type()
  40. @pytest.mark.parametrize(
  41. "attr, expect",
  42. [
  43. (
  44. {
  45. "task_type": TEST_DATABASE_TASK_TYPE,
  46. "name": "test-task-params",
  47. "datasource_name": TEST_DATABASE_DATASOURCE_NAME,
  48. "sql": TEST_DATABASE_SQL,
  49. },
  50. {
  51. "type": "MYSQL",
  52. "datasource": 1,
  53. "sql": TEST_DATABASE_SQL,
  54. "localParams": [],
  55. "resourceList": [],
  56. "dependence": {},
  57. "waitStartTimeout": {},
  58. "conditionResult": {"successNode": [""], "failedNode": [""]},
  59. },
  60. )
  61. ],
  62. )
  63. @patch(
  64. "pydolphinscheduler.core.task.Task.gen_code_and_version",
  65. return_value=(123, 1),
  66. )
  67. @patch(
  68. "pydolphinscheduler.tasks.database.Database.get_datasource_info",
  69. return_value=({"id": 1, "type": "MYSQL"}),
  70. )
  71. def test_property_task_params(mock_datasource, mock_code_version, attr, expect):
  72. """Test task database task property."""
  73. task = Database(**attr)
  74. assert expect == task.task_params
  75. @patch(
  76. "pydolphinscheduler.core.task.Task.gen_code_and_version",
  77. return_value=(123, 1),
  78. )
  79. @patch(
  80. "pydolphinscheduler.tasks.database.Database.get_datasource_info",
  81. return_value=({"id": 1, "type": "MYSQL"}),
  82. )
  83. def test_database_get_define(mock_datasource, mock_code_version):
  84. """Test task database function get_define."""
  85. name = "test_database_get_define"
  86. expect = {
  87. "code": 123,
  88. "name": name,
  89. "version": 1,
  90. "description": None,
  91. "delayTime": 0,
  92. "taskType": TEST_DATABASE_TASK_TYPE,
  93. "taskParams": {
  94. "type": "MYSQL",
  95. "datasource": 1,
  96. "sql": TEST_DATABASE_SQL,
  97. "localParams": [],
  98. "resourceList": [],
  99. "dependence": {},
  100. "conditionResult": {"successNode": [""], "failedNode": [""]},
  101. "waitStartTimeout": {},
  102. },
  103. "flag": "YES",
  104. "taskPriority": "MEDIUM",
  105. "workerGroup": "default",
  106. "failRetryTimes": 0,
  107. "failRetryInterval": 1,
  108. "timeoutFlag": "CLOSE",
  109. "timeoutNotifyStrategy": None,
  110. "timeout": 0,
  111. }
  112. task = Database(
  113. TEST_DATABASE_TASK_TYPE, name, TEST_DATABASE_DATASOURCE_NAME, TEST_DATABASE_SQL
  114. )
  115. assert task.get_define() == expect