-
Notifications
You must be signed in to change notification settings - Fork 13
/
Copy pathteam_access.py
133 lines (107 loc) · 5.18 KB
/
team_access.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
"""
Module for Terraform Enterprise/Cloud Migration Worker: Team Access.
"""
from .base_worker import TFCMigratorBaseWorker
class TeamAccessWorker(TFCMigratorBaseWorker):
"""
A class to represent the worker that will migrate all team access from one
TFC/E org to another TFC/E org.
"""
_api_module_used = "team_access"
_required_entitlements = ["teams"]
def migrate(self, workspaces_map, teams_map):
"""
Function to migrate all team access from one TFC/E org to another TFC/E org.
"""
self._logger.info("Migrating team access...")
for source_workspace_id in workspaces_map:
# Set proper workspace team filters to pull team access for each
# workspace
source_workspace_team_filters = [
{
"keys": ["workspace", "id"],
"value": source_workspace_id
}
]
# Pull teams from the old workspace
source_workspace_teams = self._api_source.team_access.list(\
filters=source_workspace_team_filters)["data"]
target_workspace_id = workspaces_map[source_workspace_id]
target_workspace_team_filters = [
{
"keys": ["workspace", "id"],
"value": target_workspace_id
}
]
target_workspace_teams = self._api_target.team_access.list(\
filters=target_workspace_team_filters)["data"]
target_team_ids = [team["relationships"]["team"]["data"]["id"] \
for team in target_workspace_teams]
for source_workspace_team in source_workspace_teams:
new_target_team_id = teams_map\
[source_workspace_team["relationships"]["team"]["data"]["id"]]
if new_target_team_id in target_team_ids:
self._logger.info(\
"Team access for workspace: %s, exists (%s). Skipped.", \
new_target_team_id, target_workspace_id)
continue
new_workspace_team_payload = {
"data": {
"attributes": {
"access": source_workspace_team["attributes"]["access"]
},
"relationships": {
"workspace": {
"data": {
"type": "workspaces",
"id": target_workspace_id
}
},
"team": {
"data": {
"type": "teams",
"id": new_target_team_id
}
}
},
"type": "team-workspaces"
}
}
if source_workspace_team["attributes"]["access"] == "custom":
attributes_to_copy = [
"runs", "variables", "state-versions", "sentinel-mocks",
"workspace-locking"
]
for attr in attributes_to_copy:
new_workspace_team_payload["data"]["attributes"][attr] = \
source_workspace_team["attributes"][attr]
# Create the team workspace access map for the target workspace
self._api_target.team_access.add_team_access(new_workspace_team_payload)
self._logger.info("Team access migrated.")
def delete_all_from_target(self):
"""
Function to delete all team access from the target TFC/E org.
"""
self._logger.info("Deleting team workspace access...")
target_workspaces = self._api_target.workspaces.list()["data"]
if target_workspaces:
for target_workspace in target_workspaces:
target_workspace_id = target_workspace["id"]
target_workspace_team_filters = [
{
"keys": ["workspace", "id"],
"value": target_workspace_id
}
]
target_workspace_teams = \
self._api_target.team_access.list(filters=target_workspace_team_filters)["data"]
if target_workspace_teams:
for target_workspace_team in target_workspace_teams:
target_workspace_team_data = self._api_target.teams.show( \
target_workspace_team["relationships"]["team"]["data"]["id"])["data"]
target_workspace_team_name = \
target_workspace_team_data["attributes"]["name"]
self._logger.info("Team access: %s, for workspace: %s...", \
target_workspace_team_name, target_workspace["attributes"]["name"])
self._api_target.team_access.remove_team_access(target_workspace_team["id"])
self._logger.info("Team workspace access deleted.")