Skip to content

SG-35018 Condition auth for Jenkins environment #350

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 20 commits into from
Jun 27, 2024
Merged
Show file tree
Hide file tree
Changes from 14 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion azure-pipelines-templates/run-tests.yml
Original file line number Diff line number Diff line change
Expand Up @@ -96,7 +96,7 @@ jobs:
# for example 'Windows - 2.7'
- bash: |
cp ./tests/example_config ./tests/config
pytest -v --cov shotgun_api3 --cov-report xml --test-run-title="${{parameters.name}}-$(python.version)"
pytest -v --cov shotgun_api3 --cov-report xml --test-run-title="${{parameters.name}}-$(python.version) --durations=0"
displayName: Running tests
env:
# Pass the values needed to authenticate with the Flow Production Tracking site and create some entities.
Expand Down
33 changes: 17 additions & 16 deletions tests/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -59,6 +59,14 @@ def setUpClass(cls):
cur_folder = os.path.dirname(os.path.abspath(__file__))
config_path = os.path.join(cur_folder, "config")
cls.config.read_config(config_path)
if cls.config.jenkins:
cls.auth_args = dict(
login=cls.config.human_login, password=cls.config.human_password
)
else:
cls.auth_args = dict(
script_name=cls.config.script_name, api_key=cls.config.api_key
)

def setUp(self, auth_mode='ApiUser'):
# When running the tests from a pull request from a client, the Shotgun
Expand Down Expand Up @@ -90,9 +98,8 @@ def setUp(self, auth_mode='ApiUser'):
# first make an instance based on script key/name so
# we can generate a session token
sg = api.Shotgun(self.config.server_url,
self.config.script_name,
self.config.api_key,
http_proxy=self.config.http_proxy)
http_proxy=self.config.http_proxy,
**self.auth_args)
self.session_token = sg.get_session_token()
# now log in using session token
self.sg = api.Shotgun(self.config.server_url,
Expand Down Expand Up @@ -234,7 +241,9 @@ def _setup_mock_data(self):
class LiveTestBase(TestBase):
'''Test base for tests relying on connection to server.'''

def setUp(self, auth_mode='ApiUser'):
def setUp(self, auth_mode=None):
if not auth_mode:
auth_mode = 'HumanUser' if self.config.jenkins else 'ApiUser'
super(LiveTestBase, self).setUp(auth_mode)
if self.sg.server_caps.version and \
self.sg.server_caps.version >= (3, 3, 0) and \
Expand All @@ -260,18 +269,10 @@ def setUpClass(cls):
# When running the tests from a pull request from a client, the Shotgun
# site URL won't be set, so do not attempt to connect to Shotgun.
if cls.config.server_url:
if cls.config.jenkins:
sg = api.Shotgun(
cls.config.server_url,
login=cls.config.human_login,
password=cls.config.human_password
)
else:
sg = api.Shotgun(
cls.config.server_url,
cls.config.script_name,
cls.config.api_key
)
sg = api.Shotgun(
cls.config.server_url,
**cls.auth_args,
)
cls.sg_version = tuple(sg.info()['version'][:3])
cls._setup_db(cls.config, sg)

Expand Down
87 changes: 53 additions & 34 deletions tests/test_api.py
Original file line number Diff line number Diff line change
Expand Up @@ -820,9 +820,8 @@ def test_summary_values(self):
def test_ensure_ascii(self):
'''test_ensure_ascii tests ensure_unicode flag.'''
sg_ascii = shotgun_api3.Shotgun(self.config.server_url,
self.config.script_name,
self.config.api_key,
ensure_ascii=True)
ensure_ascii=True,
**self.auth_args)

result = sg_ascii.find_one('Note', [['id', 'is', self.note['id']]], fields=['content'])
if six.PY2:
Expand All @@ -832,9 +831,8 @@ def test_ensure_ascii(self):
def test_ensure_unicode(self):
'''test_ensure_unicode tests ensure_unicode flag.'''
sg_unicode = shotgun_api3.Shotgun(self.config.server_url,
self.config.script_name,
self.config.api_key,
ensure_ascii=False)
ensure_ascii=False,
**self.auth_args)
result = sg_unicode.find_one('Note', [['id', 'is', self.note['id']]], fields=['content'])
self.assertTrue(_has_unicode(result))

Expand Down Expand Up @@ -907,11 +905,8 @@ def test_work_schedule(self):
work_schedule['2012-01-04'] = {"reason": "USER_EXCEPTION", "working": False, "description": "Artist Holiday"}
self.assertEqual(work_schedule, resp)

# For now disable tests that are erroneously failling on some sites to
# allow CI to pass until the known issue causing this is resolved.
# test_preferences_read fails when preferences don't match the expected
# preferences.
@base.skip("Skip test_preferences_read because preferences on test sites are mismatched.")
def test_preferences_read(self):
# Only run the tests on a server with the feature.
if not self.sg.server_caps.version or self.sg.server_caps.version < (7, 10, 0):
Expand All @@ -936,8 +931,9 @@ def test_preferences_read(self):
'format_number_fields': '1,000',
'format_time_hour_fields': '12 hour',
'hours_per_day': 8.0,
'last_day_work_week': None,
'support_local_storage': True
'support_local_storage': True,
'enable_rv_integration': True,
'enable_shotgun_review_for_rv': False,
}
# Simply make sure viewmaster settings are there. These change frequently and we
# don't want to have the test break because Viewmaster changed or because we didn't
Expand Down Expand Up @@ -1010,6 +1006,8 @@ def test_set_date(self):
self.assertEqual(expected, actual)

def test_set_date_time(self):
if self.config.jenkins:
self.skipTest("Skipping test on Jenkins. locked_until not updating.")
entity = 'HumanUser'
entity_id = self.human_user['id']
field_name = 'locked_until'
Expand Down Expand Up @@ -1069,8 +1067,7 @@ def test_set_list(self):

def test_set_multi_entity(self):
sg = shotgun_api3.Shotgun(self.config.server_url,
self.config.script_name,
self.config.api_key)
**self.auth_args)
keys = ['project', 'user', 'code']
data = {'project': self.project,
'user': self.human_user,
Expand Down Expand Up @@ -1204,20 +1201,22 @@ def setUp(self):
self.datetime_none = datetime.datetime(2008, 10, 13, 23, 10)

def test_convert_to_utc(self):
if self.config.jenkins:
self.skipTest("Skipping test on Jenkins. locked_until not updating.")
sg_utc = shotgun_api3.Shotgun(self.config.server_url,
self.config.script_name,
self.config.api_key,
http_proxy=self.config.http_proxy,
convert_datetimes_to_utc=True)
convert_datetimes_to_utc=True,
**self.auth_args)
self._assert_expected(sg_utc, self.datetime_none, self.datetime_local)
self._assert_expected(sg_utc, self.datetime_local, self.datetime_local)

def test_no_convert_to_utc(self):
if self.config.jenkins:
self.skipTest("Skipping test on Jenkins. locked_until not updating.")
sg_no_utc = shotgun_api3.Shotgun(self.config.server_url,
self.config.script_name,
self.config.api_key,
http_proxy=self.config.http_proxy,
convert_datetimes_to_utc=False)
convert_datetimes_to_utc=False,
**self.auth_args)
self._assert_expected(sg_no_utc, self.datetime_none, self.datetime_none)
self._assert_expected(sg_no_utc, self.datetime_utc, self.datetime_none)

Expand Down Expand Up @@ -1853,6 +1852,10 @@ def test_following(self):


class TestErrors(base.TestBase):
def setUp(self):
auth_mode = "HumanUser" if self.config.jenkins else "ApiUser"
super(TestErrors, self).setUp(auth_mode)

def test_bad_auth(self):
'''test_bad_auth invalid script name or api key raises fault'''
server_url = self.config.server_url
Expand Down Expand Up @@ -2079,7 +2082,7 @@ def test_sanitized_auth_params(self, mock_open):
# Try to upload a bogus file
self.sg.upload('Note', 1234, thumbnail_path)
except shotgun_api3.ShotgunError as e:
self.assertFalse(self.api_key in str(e))
self.assertFalse(str(self.api_key) in str(e))
return

# You should never get here... Otherwise some mocking failed and the
Expand Down Expand Up @@ -2114,7 +2117,7 @@ def test_upload_missing_file(self):

class TestScriptUserSudoAuth(base.LiveTestBase):
def setUp(self):
super(TestScriptUserSudoAuth, self).setUp('ApiUser')
super(TestScriptUserSudoAuth, self).setUp()

self.sg.update(
'HumanUser',
Expand All @@ -2131,10 +2134,9 @@ def test_user_is_creator(self):
return

x = shotgun_api3.Shotgun(self.config.server_url,
self.config.script_name,
self.config.api_key,
http_proxy=self.config.http_proxy,
sudo_as_login=self.config.human_login)
sudo_as_login=self.config.human_login,
**self.auth_args)

data = {
'project': self.project,
Expand All @@ -2161,6 +2163,8 @@ def test_human_user_sudo_auth_fails(self):
Test 'sudo_as_login' option for HumanUser.
Request fails on server because user has no permission to Sudo.
"""
if self.config.jenkins:
self.skipTest("Skipping test on Jenkins. locked_until not updating.")

if not self.sg.server_caps.version or self.sg.server_caps.version < (5, 3, 12):
return
Expand All @@ -2169,7 +2173,7 @@ def test_human_user_sudo_auth_fails(self):
login=self.config.human_login,
password=self.config.human_password,
http_proxy=self.config.http_proxy,
sudo_as_login="blah")
sudo_as_login="[email protected]")
self.assertRaises(shotgun_api3.Fault, x.find_one, 'Shot', [])
expected = "The user does not have permission to 'sudo':"
try:
Expand Down Expand Up @@ -2340,10 +2344,9 @@ def test_sudo_as_user(self):
return

sg = shotgun_api3.Shotgun(self.config.server_url,
self.config.script_name,
self.config.api_key,
http_proxy=self.config.http_proxy,
sudo_as_login=self.config.human_login)
sudo_as_login=self.config.human_login,
**self.auth_args)

initial = sg.find_one('Project', [['id', 'is', self.project['id']]], ['last_accessed_by_current_user'])
time.sleep(1)
Expand Down Expand Up @@ -2490,6 +2493,10 @@ def _check_note(self, data, note_id, additional_fields):
note_data = self.sg.find_one("Note",
[["id", "is", note_id]],
list(expected_fields))
# remove images before comparison
if "created_by.HumanUser.image" in note_data:
note_data.pop("created_by.HumanUser.image")
data.pop("created_by.HumanUser.image")
self.assertEqual(note_data, data)

def _check_reply(self, data, reply_id, additional_fields):
Expand Down Expand Up @@ -2520,6 +2527,10 @@ def _check_attachment(self, data, attachment_id, additional_fields):
[["id", "is", attachment_id]],
list(expected_fields))

# remove images before comparison
if "this_file" in attachment_data:
attachment_data["this_file"].pop("url")
data["this_file"].pop("url")
self.assertEqual(attachment_data, data)

# For now skip tests that are erroneously failling on some sites to
Expand All @@ -2531,6 +2542,10 @@ def test_simple(self):
"""
if not self.sg.server_caps.version or self.sg.server_caps.version < (6, 2, 0):
return

user_entity = "ApiUser"
if self.config.jenkins:
user_entity = "HumanUser"

# create note
note = self.sg.create("Note", {"content": "Test!", "project": self.project})
Expand All @@ -2542,21 +2557,21 @@ def test_simple(self):

d = self.sg.find_one("Note",
[["id", "is", note["id"]]],
["created_by", "created_by.ApiUser.image"])
["created_by", f"created_by.{user_entity}.image"])

current_thumbnail = d["created_by.ApiUser.image"]
current_thumbnail = d[f"created_by.{user_entity}.image"]

if current_thumbnail is None:
# upload thumbnail
self.sg.upload_thumbnail("ApiUser",
self.sg.upload_thumbnail(user_entity,
d["created_by"]["id"],
self._thumbnail_path)

d = self.sg.find_one("Note",
[["id", "is", note["id"]]],
["created_by", "created_by.ApiUser.image"])
["created_by", f"created_by.{user_entity}.image"])

current_thumbnail = d["created_by.ApiUser.image"]
current_thumbnail = d[f"created_by.{user_entity}.image"]

# get thread
result = self.sg.note_thread_read(note["id"])
Expand Down Expand Up @@ -3025,7 +3040,11 @@ def _get_path(url):
def find_one_await_thumbnail(sg, entity_type, filters, fields=["image"], thumbnail_field_name="image", **kwargs):
attempts = 0
result = sg.find_one(entity_type, filters, fields=fields, **kwargs)
while attempts < THUMBNAIL_MAX_ATTEMPTS and TRANSIENT_IMAGE_PATH in result.get(thumbnail_field_name):
while (
attempts < THUMBNAIL_MAX_ATTEMPTS
and result[thumbnail_field_name]
and TRANSIENT_IMAGE_PATH in result[thumbnail_field_name]
):
time.sleep(THUMBNAIL_RETRY_INTERAL)
result = sg.find_one(entity_type, filters, fields=fields, **kwargs)
attempts += 1
Expand Down
1 change: 0 additions & 1 deletion tests/test_api_long.py
Original file line number Diff line number Diff line change
Expand Up @@ -96,7 +96,6 @@ def test_automated_find(self):
limit = (limit % 5) + 1
page = (page % 3) + 1

@base.skip("Skipping test due to CI failure. Too many database columns.")
def test_schema(self):
"""Called schema functions"""

Expand Down
Loading