Commit 246df474 authored by Georges Racinet's avatar Georges Racinet 🦑
Browse files

Heptapod: users registry now contains User instances

This solves a discrepancy that has plagued the heptapod-tests
code base for long.

In the beginning, users where represented as simple dicts, themselves
kept in a registry: `user` attribute of the `Heptapod` instance.
Then, more advanced manipulation needs pushed to the creation of
the `User` class, but the user registry wasn't updated at the time.

The tricky part, explaining why it wasn't made right away is that
we often start with a partial representation of the user: for instance
not knowing its numeric id.

Over time some of the test code has been updated to use higher-level
methods of the `Heptapod` class, such as `get_user_webdriver()`, in
prevision for the present move, but still, the registry values remained
at heart just dicts, while lots of needed logic is implemented in
the `User` class.

Now we'll have `User` instances in the registry. That still leaves us
with some illogical aspects, like lazy webdriver and token initialization
indirected through the `Heptapod` class, instead of being transparent
properties `User`. But from now on, improving on this should be incremental
and involve identifiers easy to search in the code base.

branch : stable
parent cc7476e40e98
Pipeline #17958 passed with stages
in 23 minutes
......@@ -175,7 +175,7 @@ def test_cli_404(test_project):
heptapod = test_project.heptapod
user = 'test_basic'
basic_user_creds = (user, heptapod.users[user]['password'])
basic_user_creds = (user, heptapod.users[user].password)
resp = requests.get(heptapod.url + '/no/such/project?cmd=capabilities',
......@@ -33,7 +33,7 @@ class Group:
if parent is not None:
data['parent_id'] =
headers = {'Private-Token': heptapod.users[user_name]['token']}
headers = {'Private-Token': heptapod.users[user_name].token}
resp = + cls.api_uri,
......@@ -57,18 +57,21 @@ class Group:
grp.full_path = info['full_path']
return grp
def api_get(self):
def private_token(self):
"""Return a token strong enough for all operations.
Namely, using a token for an Owner, if there's a known one or
an Administrator token
user_name = self.owner_name if self.owner_name is not None else 'root'
headers = {'Private-Token':
return requests.get(self.api_url, headers=headers)
return {'Private-Token': self.heptapod.users[user_name].token}
def api_get(self):
return requests.get(self.api_url, headers=self.private_token())
def api_post(self, subpath='', **params):
user_name = self.owner_name if self.owner_name is not None else 'root'
headers = {'Private-Token':
return'/'.join((self.api_url, subpath)),
def api_add_member(self, user, access_level):
......@@ -78,7 +81,7 @@ class Group:
def api_search(cls, heptapod, group_name, user_name='root'):
headers = {'Private-Token': heptapod.users[user_name]['token']}
headers = {'Private-Token': heptapod.users[user_name].token}
resp = requests.get(heptapod.url + cls.api_uri,
......@@ -93,10 +96,7 @@ class Group:
return self.heptapod.url + self.api_uri + '/%d' %
def api_delete(self):
user_name = self.owner_name if self.owner_name is not None else 'root'
headers = {'Private-Token':
resp = requests.delete(self.api_url, headers=headers)
resp = requests.delete(self.api_url, headers=self.private_token())
assert resp.status_code in (202, 204)
def fs_path(self):
......@@ -107,11 +107,11 @@ class Heptapod:
def root_token_headers(self):
return {'Private-Token': self.users['root']['token']}
return {'Private-Token': self.users['root'].token}
def basic_user_token_headers(self):
return {'Private-Token': self.users['test_basic']['token']}
return {'Private-Token': self.users['test_basic'].token}
def run_shell(self, command, **kw):
exit_code, output = self.execute(command, **kw)
......@@ -126,22 +126,7 @@ class Heptapod:
def get_user(self, name):
"""Return a :class:`User` instance, or `None`."""
info = self.users.get(name)
if info is None:
return None
user_id = info.get('id')
password = info.get('password')
if user_id is None:
"Searching for known user %r because its id is unknown.",
user =, name)
user.password = password
return user
return User(heptapod=self, id=user_id, name=name, password=password)
return self.users.get(name)
def new_webdriver(self):
options = selenium.webdriver.ChromeOptions()
......@@ -157,8 +142,8 @@ class Heptapod:
return selenium.webdriver.Chrome(options=options)
def get_user_webdriver(self, user_name):
info = self.users[user_name]
driver = info.get('webdriver')
user = self.users[user_name]
driver = user.webdriver
if driver is not None:
return driver
......@@ -169,13 +154,12 @@ class Heptapod:
driver = self.new_webdriver()
# guaranteeing driver to be available for teardown
# as soon as created
info['webdriver'] = driver
session.login_as_root(driver, self, info['password'])
session.login_as_root(driver, self, user.password)
# TODO should init webdriver from here and store it before login
# attempt as well
driver = session.make_webdriver(self, user_name, info['password'])
info['webdriver'] = driver
driver = session.make_webdriver(self, user_name, user.password)
user.webdriver = driver
return driver
def wait_startup(self, first_response_timeout=INITIAL_TIMEOUT,
......@@ -235,12 +219,20 @@ class Heptapod:
"or initialized", path)
for name, info in cached['users'].items():
self.users[name] = dict(name=name,
if 'id' not in info:
logger.warning("Cache file %r is from an earlier version "
"of heptapod-tests. "
"Removing to start afresh.", path)
return self.load_instance_cache()
self.users[name] = User(heptapod=self,
def update_instance_cache(self):
users = {name: dict(token=info['token'])
for name, info in self.users.items()}
users = { dict(token=user.token,
for user in self.users.values()}
with open(self.instance_cache_file(), 'w') as cachef:
json.dump(dict(users=users), cachef)
......@@ -264,7 +256,11 @@ class Heptapod:"Preparing root user.")
start = time.time()
root = User.init_root(self, root_password)
root = self.users.get('root')
if root is None:
root = User.init_root(self, root_password)
root.password = root_password
root.ensure_private_token()"Preparing application settings.")
......@@ -354,28 +350,28 @@ class Heptapod:
Also makes sure the keys are actually usable (perms)
ssh_dir = Path(__file__).parent.parent / 'data' / 'ssh'
for name, info in self.users.items():
base_fname = 'id_rsa_heptapod_' + name
for user in self.users.values():
base_fname = 'id_rsa_heptapod_' +
priv = ssh_dir / base_fname
pub = ssh_dir / (base_fname + '.pub')
# VCSes tend not to preserve non-executable perm bits
info['ssh'] = dict(priv=str(priv), pub=pub.read_text())
user.ssh = dict(priv=str(priv), pub=pub.read_text())
def upload_ssh_pub_keys(self):
"""Upload SSH public keys for all users to Heptapod."""
# it's really time to put the actual user object in our `self.users`
for name, info in self.users.items():
user =, name)
for user in self.users.values():
def close_webdrivers(self):
for user in self.users.values():
driver = user.pop('webdriver', None)
driver = user.webdriver
if driver is not None:"Closing webdriver for user %r", user['name'])"Closing webdriver for user %r",
user.webdriver = None
def close(self):
if self.dead is not False:
......@@ -445,7 +441,7 @@ class Heptapod:
# Worst case scenario, we lost all our data. We need to
# reprepare the server for subsequent tests
def set_vcs_types_settings(self, vcs_types):
......@@ -35,19 +35,20 @@ def finalize_password(driver, heptapod, user, password, current_password=None):
def sign_in_page_login(driver, heptapod, user, password=None):
def sign_in_page_login(driver, heptapod, user_name, password=None):
"""Perform login as user, with webdriver already on the signin page.
If password is not specified, it is read from the heptapod object.
If it is, it'll be stored in the heptapod object
elem = driver.find_element_by_name('user[login]')
elem = driver.find_element_by_name('user[password]')
user = heptapod.users[user_name]
if password is None:
password = heptapod.users[user]['password']
password = user.password
heptapod.users.setdefault(user, {})['password'] = password
user.password = password
......@@ -86,14 +87,17 @@ def ensure_user(heptapod, name, password=None, fullname=None):
if password is None:
password = default_password(password)
user =, name)
user = heptapod.users.get(name)
if user is None:
user =, name)
if user is None:
user = User.create(heptapod, name,
password=password, fullname=fullname)
user.password = password
user.password = password
return user
......@@ -18,6 +18,8 @@ class User(object):
id = attr.ib()
heptapod = attr.ib(repr=False)
password = attr.ib(default=None, repr=False)
token = attr.ib(default=None, repr=False)
webdriver = attr.ib(default=None, repr=False)
_python_gitlab_client = attr.ib(default=None, repr=False)
......@@ -75,15 +77,7 @@ class User(object):
return user
def store_in_heptapod(self):
as_hepta_dict = self.heptapod.users.setdefault(, {})
# password may be already known, don't overwrite it
# (this will be simpler when Heptapod stores User instances
if self.password is not None:
as_hepta_dict['password'] = self.password
self.heptapod.users[] = self
def api_request(self, method, subpath=None, **data):
"""Perform a request on user API as root"""
......@@ -164,33 +158,13 @@ class User(object):
'-o', 'IdentitiesOnly=yes',
'-o', 'NoHostAuthenticationForLocalhost=yes ',
'-o', 'StrictHostKeyChecking=no',
'-i', self.heptapod.users[]['ssh']['priv']))
def token(self):
"""Separate method for future changes.
At some point we'll stop storing this on the Heptapod object
return self.heptapod.users[].get('token')
'-i', self.ssh['priv']))
def token_headers(self):
"""Return a *new* headers dict suitable for authenticated API acces.
return {'Private-Token': self.token}
def token(self, value):
self.heptapod.users.setdefault(, {})['token'] = value
def webdriver(self):
"""Separate method for future changes.
At some point we'll stop storing this on the Heptapod object
return self.heptapod.get_user_webdriver(
def basic_auth_url(self):
url = self.heptapod.parsed_url
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment