mirror of
https://github.com/ansible/awx.git
synced 2026-02-23 22:16:00 -03:30
333 lines
12 KiB
Python
333 lines
12 KiB
Python
from django.contrib.auth.models import User
|
|
|
|
from awx.main.models import (
|
|
Organization,
|
|
Project,
|
|
Team,
|
|
NotificationTemplate,
|
|
Credential,
|
|
Inventory,
|
|
Job,
|
|
Label,
|
|
)
|
|
|
|
from .objects import (
|
|
generate_objects,
|
|
generate_role_objects,
|
|
_Mapped,
|
|
)
|
|
|
|
from .fixtures import (
|
|
mk_organization,
|
|
mk_team,
|
|
mk_user,
|
|
mk_job_template,
|
|
mk_job,
|
|
mk_credential,
|
|
mk_inventory,
|
|
mk_project,
|
|
mk_label,
|
|
mk_notification_template,
|
|
)
|
|
|
|
|
|
def apply_roles(roles, objects, persisted):
|
|
'''apply_roles evaluates a list of Role relationships represented as strings.
|
|
The format of this string is 'role:[user|role]'. When a user is provided, they will be
|
|
made a member of the role on the LHS. When a role is provided that role will be added to
|
|
the children of the role on the LHS.
|
|
|
|
This function assumes that objects is a dictionary that contains a unique set of key to value
|
|
mappings for all possible "Role objects". See the example below:
|
|
|
|
Mapping Users
|
|
-------------
|
|
roles = ['org1.admin_role:user1', 'team1.admin_role:user1']
|
|
objects = {'org1': Organization, 'team1': Team, 'user1': User]
|
|
|
|
Mapping Roles
|
|
-------------
|
|
roles = ['org1.admin_role:team1.admin_role']
|
|
objects = {'org1': Organization, 'team1': Team}
|
|
|
|
Invalid Mapping
|
|
---------------
|
|
roles = ['org1.admin_role:team1.admin_role']
|
|
objects = {'org1': Organization', 'user1': User} # Exception, no team1 entry
|
|
'''
|
|
if roles is None:
|
|
return None
|
|
|
|
if not persisted:
|
|
raise RuntimeError('roles can not be used when persisted=False')
|
|
|
|
for role in roles:
|
|
obj_role, sep, member_role = role.partition(':')
|
|
if not member_role:
|
|
raise RuntimeError('you must provide an assignment role, got None')
|
|
|
|
obj_str, o_role_str = obj_role.split('.')
|
|
member_str, m_sep, m_role_str = member_role.partition('.')
|
|
|
|
obj = objects[obj_str]
|
|
obj_role = getattr(obj, o_role_str)
|
|
|
|
member = objects[member_str]
|
|
if m_role_str:
|
|
if hasattr(member, m_role_str):
|
|
member_role = getattr(member, m_role_str)
|
|
obj_role.children.add(member_role)
|
|
else:
|
|
raise RuntimeError('unable to find {} role for {}'.format(m_role_str, member_str))
|
|
else:
|
|
if type(member) is User:
|
|
obj_role.members.add(member)
|
|
else:
|
|
raise RuntimeError('unable to add non-user {} for members list of {}'.format(member_str, obj_str))
|
|
|
|
def generate_users(organization, teams, superuser, persisted, **kwargs):
|
|
'''generate_users evaluates a mixed list of User objects and strings.
|
|
If a string is encountered a user with that username is created and added to the lookup dict.
|
|
If a User object is encountered the User.username is used as a key for the lookup dict.
|
|
|
|
A short hand for assigning a user to a team is available in the following format: "team_name:username".
|
|
If a string in that format is encounted an attempt to lookup the team by the key team_name from the teams
|
|
argumnent is made, a KeyError will be thrown if the team does not exist in the dict. The teams argument should
|
|
be a dict of {Team.name:Team}
|
|
'''
|
|
users = {}
|
|
key = 'superusers' if superuser else 'users'
|
|
if key in kwargs and kwargs.get(key) is not None:
|
|
for u in kwargs[key]:
|
|
if type(u) is User:
|
|
users[u.username] = u
|
|
else:
|
|
p1, sep, p2 = u.partition(':')
|
|
if p2:
|
|
t = teams[p1]
|
|
users[p2] = mk_user(p2, organization=organization, team=t, is_superuser=superuser, persisted=persisted)
|
|
else:
|
|
users[p1] = mk_user(p1, organization=organization, team=None, is_superuser=superuser, persisted=persisted)
|
|
return users
|
|
|
|
def generate_teams(organization, persisted, **kwargs):
|
|
'''generate_teams evalutes a mixed list of Team objects and strings.
|
|
If a string is encountered a team with that string name is created and added to the lookup dict.
|
|
If a Team object is encounted the Team.name is used as a key for the lookup dict.
|
|
'''
|
|
teams = {}
|
|
if 'teams' in kwargs and kwargs.get('teams') is not None:
|
|
for t in kwargs['teams']:
|
|
if type(t) is Team:
|
|
teams[t.name] = t
|
|
else:
|
|
teams[t] = mk_team(t, organization=organization, persisted=persisted)
|
|
return teams
|
|
|
|
def create_survey_spec(variables=None, default_type='integer', required=True):
|
|
'''
|
|
Returns a valid survey spec for a job template, based on the input
|
|
argument specifying variable name(s)
|
|
'''
|
|
if isinstance(variables, list):
|
|
name = "%s survey" % variables[0]
|
|
description = "A survey that starts with %s." % variables[0]
|
|
vars_list = variables
|
|
else:
|
|
name = "%s survey" % variables
|
|
description = "A survey about %s." % variables
|
|
vars_list = [variables]
|
|
|
|
spec = []
|
|
index = 0
|
|
for var in vars_list:
|
|
spec_item = {}
|
|
spec_item['index'] = index
|
|
index += 1
|
|
spec_item['required'] = required
|
|
spec_item['choices'] = ''
|
|
spec_item['type'] = default_type
|
|
if isinstance(var, dict):
|
|
spec_item.update(var)
|
|
var_name = spec_item.get('variable', 'variable')
|
|
else:
|
|
var_name = var
|
|
spec_item.setdefault('variable', var_name)
|
|
spec_item.setdefault('question_name', "Enter a value for %s." % var_name)
|
|
spec_item.setdefault('question_description', "A question about %s." % var_name)
|
|
if spec_item['type'] == 'integer':
|
|
spec_item.setdefault('default', 0)
|
|
spec_item.setdefault('max', spec_item['default'] + 100)
|
|
spec_item.setdefault('min', spec_item['default'] - 100)
|
|
else:
|
|
spec_item.setdefault('default', '')
|
|
spec.append(spec_item)
|
|
|
|
survey_spec = {}
|
|
survey_spec['spec'] = spec
|
|
survey_spec['name'] = name
|
|
survey_spec['description'] = description
|
|
return survey_spec
|
|
|
|
|
|
# create methods are intended to be called directly as needed
|
|
# or encapsulated by specific factory fixtures in a conftest
|
|
#
|
|
|
|
def create_job_template(name, roles=None, persisted=True, **kwargs):
|
|
Objects = generate_objects(["job_template", "jobs",
|
|
"organization",
|
|
"inventory",
|
|
"project",
|
|
"credential",
|
|
"job_type",
|
|
"survey",], kwargs)
|
|
|
|
org = None
|
|
proj = None
|
|
inv = None
|
|
cred = None
|
|
spec = None
|
|
jobs = {}
|
|
job_type = kwargs.get('job_type', 'run')
|
|
extra_vars = kwargs.get('extra_vars', '')
|
|
|
|
if 'organization' in kwargs:
|
|
org = kwargs['organization']
|
|
if type(org) is not Organization:
|
|
org = mk_organization(org, '%s-desc'.format(org), persisted=persisted)
|
|
|
|
if 'credential' in kwargs:
|
|
cred = kwargs['credential']
|
|
if type(cred) is not Credential:
|
|
cred = mk_credential(cred, persisted=persisted)
|
|
|
|
if 'project' in kwargs:
|
|
proj = kwargs['project']
|
|
if type(proj) is not Project:
|
|
proj = mk_project(proj, organization=org, persisted=persisted)
|
|
|
|
if 'inventory' in kwargs:
|
|
inv = kwargs['inventory']
|
|
if type(inv) is not Inventory:
|
|
inv = mk_inventory(inv, organization=org, persisted=persisted)
|
|
|
|
if 'survey' in kwargs:
|
|
spec = create_survey_spec(kwargs['survey'])
|
|
|
|
jt = mk_job_template(name, project=proj,
|
|
inventory=inv, credential=cred,
|
|
job_type=job_type, spec=spec, extra_vars=extra_vars,
|
|
persisted=persisted)
|
|
|
|
if 'jobs' in kwargs:
|
|
for i in kwargs['jobs']:
|
|
if type(i) is Job:
|
|
jobs[i.pk] = i
|
|
else:
|
|
# Fill in default survey answers
|
|
job_extra_vars = {}
|
|
for question in spec['spec']:
|
|
job_extra_vars[question['variable']] = question['default']
|
|
jobs[i] = mk_job(job_template=jt, project=proj, inventory=inv, credential=cred,
|
|
extra_vars=job_extra_vars,
|
|
job_type=job_type, persisted=persisted)
|
|
|
|
role_objects = generate_role_objects([org, proj, inv, cred])
|
|
apply_roles(roles, role_objects, persisted)
|
|
|
|
return Objects(job_template=jt,
|
|
jobs=jobs,
|
|
project=proj,
|
|
inventory=inv,
|
|
credential=cred,
|
|
job_type=job_type,
|
|
organization=org,
|
|
survey=spec,)
|
|
|
|
def create_organization(name, roles=None, persisted=True, **kwargs):
|
|
Objects = generate_objects(["organization",
|
|
"teams", "users",
|
|
"superusers",
|
|
"projects",
|
|
"labels",
|
|
"notification_templates",
|
|
"inventories",], kwargs)
|
|
|
|
projects = {}
|
|
inventories = {}
|
|
labels = {}
|
|
notification_templates = {}
|
|
|
|
org = mk_organization(name, '%s-desc'.format(name), persisted=persisted)
|
|
|
|
if 'inventories' in kwargs:
|
|
for i in kwargs['inventories']:
|
|
if type(i) is Inventory:
|
|
inventories[i.name] = i
|
|
else:
|
|
inventories[i] = mk_inventory(i, organization=org, persisted=persisted)
|
|
|
|
if 'projects' in kwargs:
|
|
for p in kwargs['projects']:
|
|
if type(p) is Project:
|
|
projects[p.name] = p
|
|
else:
|
|
projects[p] = mk_project(p, organization=org, persisted=persisted)
|
|
|
|
teams = generate_teams(org, persisted, teams=kwargs.get('teams'))
|
|
superusers = generate_users(org, teams, True, persisted, superusers=kwargs.get('superusers'))
|
|
users = generate_users(org, teams, False, persisted, users=kwargs.get('users'))
|
|
|
|
if 'labels' in kwargs:
|
|
for l in kwargs['labels']:
|
|
if type(l) is Label:
|
|
labels[l.name] = l
|
|
else:
|
|
labels[l] = mk_label(l, organization=org, persisted=persisted)
|
|
|
|
if 'notification_templates' in kwargs:
|
|
for nt in kwargs['notification_templates']:
|
|
if type(nt) is NotificationTemplate:
|
|
notification_templates[nt.name] = nt
|
|
else:
|
|
notification_templates[nt] = mk_notification_template(nt, organization=org, persisted=persisted)
|
|
|
|
role_objects = generate_role_objects([org, superusers, users, teams, projects, labels, notification_templates])
|
|
apply_roles(roles, role_objects, persisted)
|
|
return Objects(organization=org,
|
|
superusers=_Mapped(superusers),
|
|
users=_Mapped(users),
|
|
teams=_Mapped(teams),
|
|
projects=_Mapped(projects),
|
|
labels=_Mapped(labels),
|
|
notification_templates=_Mapped(notification_templates),
|
|
inventories=_Mapped(inventories))
|
|
|
|
def create_notification_template(name, roles=None, persisted=True, **kwargs):
|
|
Objects = generate_objects(["notification_template",
|
|
"organization",
|
|
"users",
|
|
"superusers",
|
|
"teams",], kwargs)
|
|
|
|
organization = None
|
|
|
|
if 'organization' in kwargs:
|
|
org = kwargs['organization']
|
|
organization = mk_organization(org, '{}-desc'.format(org), persisted=persisted)
|
|
|
|
notification_template = mk_notification_template(name, organization=organization, persisted=persisted)
|
|
|
|
teams = generate_teams(organization, persisted, teams=kwargs.get('teams'))
|
|
superusers = generate_users(organization, teams, True, persisted, superusers=kwargs.get('superusers'))
|
|
users = generate_users(organization, teams, False, persisted, users=kwargs.get('users'))
|
|
|
|
role_objects = generate_role_objects([organization, notification_template])
|
|
apply_roles(roles, role_objects, persisted)
|
|
return Objects(notification_template=notification_template,
|
|
organization=organization,
|
|
users=_Mapped(users),
|
|
superusers=_Mapped(superusers),
|
|
teams=teams)
|