-
Notifications
You must be signed in to change notification settings - Fork 1.7k
/
conftest.py
150 lines (128 loc) · 3.69 KB
/
conftest.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
import datetime
import pytest
from pytest_mock import MockFixture
from common.utils.const import WorkflowStatus
from sql.models import (
Instance,
ResourceGroup,
SqlWorkflow,
SqlWorkflowContent,
QueryPrivilegesApply,
ArchiveConfig,
InstanceTag,
)
from common.config import SysConfig
from sql.utils.workflow_audit import AuditV2, AuditSetting
@pytest.fixture
def normal_user(django_user_model):
user = django_user_model.objects.create(
username="test_user", display="中文显示", is_active=True
)
yield user
user.delete()
@pytest.fixture
def super_user(django_user_model):
user = django_user_model.objects.create(
username="super_user", display="超级用户", is_active=True, is_superuser=True
)
yield user
user.delete()
@pytest.fixture
def db_instance(db):
ins = Instance.objects.create(
instance_name="some_ins",
type="slave",
db_type="mysql",
host="some_host",
port=3306,
user="ins_user",
password="some_str",
)
yield ins
ins.delete()
@pytest.fixture
def resource_group(db) -> ResourceGroup:
res_group = ResourceGroup.objects.create(group_id=1, group_name="group_name")
yield res_group
res_group.delete()
@pytest.fixture
def sql_workflow(db_instance):
wf = SqlWorkflow.objects.create(
workflow_name="some_name",
group_id=1,
group_name="g1",
engineer_display="",
audit_auth_groups="some_audit_group",
create_time=datetime.datetime.now(),
status="workflow_timingtask",
is_backup=True,
instance=db_instance,
db_name="some_db",
syntax_type=1,
)
wf_content = SqlWorkflowContent.objects.create(
workflow=wf, sql_content="some_sql", execute_result=""
)
yield wf, wf_content
wf.delete()
wf_content.delete()
@pytest.fixture
def sql_query_apply(db_instance):
tomorrow = datetime.datetime.today() + datetime.timedelta(days=1)
query_apply_1 = QueryPrivilegesApply.objects.create(
group_id=1,
group_name="some_name",
title="some_title1",
user_name="some_user",
instance=db_instance,
db_list="some_db,some_db2",
limit_num=100,
valid_date=tomorrow,
priv_type=1,
status=0,
audit_auth_groups="1",
)
yield query_apply_1
query_apply_1.delete()
@pytest.fixture
def archive_apply(db_instance, resource_group):
archive_apply_1 = ArchiveConfig.objects.create(
title="title",
resource_group=resource_group,
audit_auth_groups="",
src_instance=db_instance,
src_db_name="src_db_name",
src_table_name="src_table_name",
dest_instance=db_instance,
dest_db_name="src_db_name",
dest_table_name="src_table_name",
condition="1=1",
mode="file",
no_delete=True,
sleep=1,
status=WorkflowStatus.WAITING,
state=False,
user_name="some_user",
user_display="display",
)
yield archive_apply_1
archive_apply_1.delete()
@pytest.fixture
def setup_sys_config(db):
sys_config = SysConfig()
yield sys_config
sys_config.purge()
@pytest.fixture
def fake_generate_audit_setting(mocker: MockFixture):
mock_generate_audit_setting = mocker.patch.object(AuditV2, "generate_audit_setting")
fake_audit_setting = AuditSetting(
auto_pass=False,
audit_auth_groups=[123],
)
mock_generate_audit_setting.return_value = fake_audit_setting
yield mock_generate_audit_setting
@pytest.fixture
def instance_tag(db):
tag = InstanceTag.objects.create(tag_code="test_tag", tag_name="测试标签")
yield tag
tag.delete()