def setUp(self):

super(TestShotgunApi, self).setUp()

# give note unicode content

self.sg.update('Note', self.note['id'], {'content': u'La Pe\xf1a'})

def test_info(self):

"""Called info"""

# TODO do more to check results

self.sg.info()

def test_server_dates(self):

"""Pass datetimes to the server"""

# TODO check results

t = {'project': self.project,

'start_date': datetime.date.today()}

self.sg.create('Task', t, ['content', 'sg_status_list'])

def test_batch(self):

"""Batched create, update, delete"""

requests = [

{

"request_type": "create",

"entity_type": "Shot",

"data": {

"code": "New Shot 5",

"project": self.project

}

},

{

"request_type": "update",

"entity_type": "Shot",

"entity_id": self.shot['id'],

"data": {

"code": "Changed 1"

}

}

]

new_shot, updated_shot = self.sg.batch(requests)

self.assertEqual(self.shot['id'], updated_shot["id"])

self.assertTrue(new_shot.get("id"))

new_shot_id = new_shot["id"]

requests = [

{

"request_type": "delete",

"entity_type": "Shot",

"entity_id": new_shot_id

},

{

"request_type": "update",

"entity_type": "Shot",

"entity_id": self.shot['id'],

"data": {"code": self.shot['code']}

}

]

result = self.sg.batch(requests)[0]

self.assertEqual(True, result)

def test_empty_batch(self):

"""Empty list sent to .batch()"""

result = self.sg.batch([])

self.assertEqual([], result)

def test_create_update_delete(self):

"""Called create, update, delete, revive"""

data = {

'project': self.project,

'code': 'JohnnyApple_Design01_FaceFinal',

'description': 'fixed rig per director final notes',

'sg_status_list': 'rev',

'entity': self.asset,

'user': self.human_user

}

version = self.sg.create("Version", data, return_fields=["id"])

self.assertTrue(isinstance(version, dict))

self.assertTrue("id" in version)

# TODO check results more thoroughly

# TODO: test returned fields are requested fields

data = data = {

"description": "updated test"

}

version = self.sg.update("Version", version["id"], data)

self.assertTrue(isinstance(version, dict))

self.assertTrue("id" in version)

rv = self.sg.delete("Version", version["id"])

self.assertEqual(True, rv)

rv = self.sg.delete("Version", version["id"])

self.assertEqual(False, rv)

rv = self.sg.revive("Version", version["id"])

self.assertEqual(True, rv)

rv = self.sg.revive("Version", version["id"])

self.assertEqual(False, rv)

def test_last_accessed(self):

page = self.sg.find('Page', [], fields=['last_accessed'], limit=1)

self.assertEqual("Page", page[0]['type'])

self.assertEqual(datetime.datetime, type(page[0]['last_accessed']))

def test_get_session_token(self):

"""Got session UUID"""

# TODO test results

rv = self.sg.get_session_token()

self.assertTrue(rv)

def test_upload_download(self):

"""Upload and download an attachment tests"""

# upload / download only works against a live server because it does

# not use the standard http interface

if 'localhost' in self.server_url:

print("upload / down tests skipped for localhost")

return

this_dir, _ = os.path.split(__file__)

path = os.path.abspath(os.path.expanduser(

os.path.join(this_dir, "sg_logo.jpg")))

size = os.stat(path).st_size

attach_id = self.sg.upload("Ticket",

self.ticket['id'], path, 'attachments',

tag_list="monkeys, everywhere, send, help")

# test download with attachment_id

attach_file = self.sg.download_attachment(attach_id)

self.assertTrue(attach_file is not None)

self.assertEqual(size, len(attach_file))

orig_file = open(path, "rb").read()

self.assertEqual(orig_file, attach_file)

# test download with attachment_id as keyword

attach_file = self.sg.download_attachment(attachment_id=attach_id)

self.assertTrue(attach_file is not None)

self.assertEqual(size, len(attach_file))

orig_file = open(path, "rb").read()

self.assertEqual(orig_file, attach_file)

# test download with attachment_id (write to disk)

file_path = os.path.join(os.path.dirname(os.path.realpath(__file__)), "sg_logo_download.jpg")

result = self.sg.download_attachment(attach_id, file_path=file_path)

self.assertEqual(result, file_path)

# On windows read may not read to end of file unless opened 'rb'

fp = open(file_path, 'rb')

attach_file = fp.read()

fp.close()

self.assertEqual(size, len(attach_file))

self.assertEqual(orig_file, attach_file)

# test download with attachment hash

ticket = self.sg.find_one('Ticket', [['id', 'is', self.ticket['id']]],

['attachments'])

# Look for the attachment we just uploaded, the attachments are not returned from latest

# to earliest.

attachment = [x for x in ticket["attachments"] if x["id"] == attach_id]

self.assertEqual(len(attachment), 1)

attachment = attachment[0]

attach_file = self.sg.download_attachment(attachment)

self.assertTrue(attach_file is not None)

self.assertEqual(size, len(attach_file))

self.assertEqual(orig_file, attach_file)

# test download with attachment hash (write to disk)

result = self.sg.download_attachment(attachment,

file_path=file_path)

self.assertEqual(result, file_path)

fp = open(file_path, 'rb')

attach_file = fp.read()

fp.close()

self.assertTrue(attach_file is not None)

self.assertEqual(size, len(attach_file))

self.assertEqual(orig_file, attach_file)

# test invalid requests

INVALID_S3_URL = "https://sg-media-usor-01.s3.amazonaws.com/ada3de3ee3873875e1dd44f2eb0882c75ae36a4a/cd31346421dbeef781e0e480f259a3d36652d7f2/IMG_0465.MOV?AWSAccessKeyId=AKIAIQGOBSVN3FSQ5QFA&Expires=1371789959&Signature=SLbzv7DuVlZ8XAoOSQQAiGpF3u8%3D" # noqa

self.assertRaises(shotgun_api3.ShotgunFileDownloadError,

self.sg.download_attachment,

{"url": INVALID_S3_URL})

INVALID_ATTACHMENT_ID = 99999999

self.assertRaises(shotgun_api3.ShotgunFileDownloadError,

self.sg.download_attachment,

INVALID_ATTACHMENT_ID)

self.assertRaises(TypeError, self.sg.download_attachment,

"/path/to/some/file.jpg")

self.assertRaises(ValueError, self.sg.download_attachment,

{"id": 123, "type": "Shot"})

self.assertRaises(TypeError, self.sg.download_attachment)

# test upload of non-ascii, unicode path

u_path = os.path.abspath(

os.path.expanduser(

glob.glob(os.path.join(six.text_type(this_dir), u'No*l.jpg'))[0]

)

)

# If this is a problem, it'll raise with a UnicodeEncodeError. We

# don't need to check the results of the upload itself -- we're

# only checking that the non-ascii string encoding doesn't trip

# us up the way it used to.

self.sg.upload(

"Ticket",

self.ticket['id'],

u_path,

'attachments',

tag_list="monkeys, everywhere, send, help"

)

# Also make sure that we can pass in a utf-8 encoded string path

# with non-ascii characters and have it work properly. This is

# primarily a concern on Windows, as it doesn't handle that

# situation as well as OS X and Linux.

self.sg.upload(

"Ticket",

self.ticket['id'],

u_path.encode("utf-8"),

'attachments',

tag_list="monkeys, everywhere, send, help"

)

if six.PY2:

# In Python2, make sure that non-utf-8 encoded paths raise when they

# can't be converted to utf-8. For Python3, we'll skip these tests

# since string encoding is handled differently.

# We need to touch the file we're going to test with first. We can't

# bundle a file with this filename in the repo due to some pip install

# problems on Windows. Note that the path below is utf-8 encoding of

# what we'll eventually encode as shift-jis.

file_path_s = os.path.join(this_dir, "./\xe3\x81\x94.shift-jis")

file_path_u = file_path_s.decode("utf-8")

with open(file_path_u if sys.platform.startswith("win") else file_path_s, "w") as fh:

fh.write("This is just a test file with some random data in it.")

self.assertRaises(

shotgun_api3.ShotgunError,

self.sg.upload,

"Ticket",

self.ticket['id'],

file_path_u.encode("shift-jis"),

'attachments',

tag_list="monkeys, everywhere, send, help"

)

# But it should work in all cases if a unicode string is used.

self.sg.upload(

"Ticket",

self.ticket['id'],

file_path_u,

'attachments',

tag_list="monkeys, everywhere, send, help"

)

# cleanup

os.remove(file_path_u)

# cleanup

os.remove(file_path)

def test_upload_thumbnail_in_create(self):

"""Upload a thumbnail via the create method"""

this_dir, _ = os.path.split(__file__)

path = os.path.abspath(os.path.expanduser(

os.path.join(this_dir, "sg_logo.jpg")))

# test thumbnail upload

data = {'image': path, 'code': 'Test Version',

'project': self.project}

new_version = self.sg.create("Version", data, return_fields=['image'])

new_version = find_one_await_thumbnail(

self.sg,

"Version",

[["id", "is", new_version["id"]]],

fields=["image", "project", "type", "id"]

)

self.assertTrue(new_version is not None)

self.assertTrue(isinstance(new_version, dict))

self.assertTrue(isinstance(new_version.get('id'), int))

self.assertEqual(new_version.get('type'), 'Version')

self.assertEqual(new_version.get('project'), self.project)

self.assertTrue(new_version.get('image') is not None)

h = Http(".cache")

thumb_resp, content = h.request(new_version.get('image'), "GET")

self.assertEqual(thumb_resp['status'], '200')

self.assertEqual(thumb_resp['content-type'], 'image/jpeg')

self.sg.delete("Version", new_version['id'])

# test filmstrip image upload

data = {'filmstrip_image': path, 'code': 'Test Version',

'project': self.project}

new_version = self.sg.create("Version", data, return_fields=['filmstrip_image'])

self.assertTrue(new_version is not None)

self.assertTrue(isinstance(new_version, dict))

self.assertTrue(isinstance(new_version.get('id'), int))

self.assertEqual(new_version.get('type'), 'Version')

self.assertEqual(new_version.get('project'), self.project)

self.assertTrue(new_version.get('filmstrip_image') is not None)

url = new_version.get('filmstrip_image')

data = self.sg.download_attachment({'url': url})

self.assertTrue(isinstance(data, six.binary_type))

self.sg.delete("Version", new_version['id'])

# end test_upload_thumbnail_in_create

def test_upload_thumbnail_for_version(self):

"""simple upload thumbnail for version test."""

this_dir, _ = os.path.split(__file__)

path = os.path.abspath(os.path.expanduser(

os.path.join(this_dir, "sg_logo.jpg")))

# upload thumbnail

thumb_id = self.sg.upload_thumbnail("Version", self.version['id'], path)

self.assertTrue(isinstance(thumb_id, int))

# check result on version

version_with_thumbnail = self.sg.find_one('Version', [['id', 'is', self.version['id']]])

version_with_thumbnail = find_one_await_thumbnail(self.sg, 'Version', [['id', 'is', self.version['id']]])

self.assertEqual(version_with_thumbnail.get('type'), 'Version')

self.assertEqual(version_with_thumbnail.get('id'), self.version['id'])

h = Http(".cache")

thumb_resp, content = h.request(version_with_thumbnail.get('image'), "GET")

self.assertEqual(thumb_resp['status'], '200')

self.assertEqual(thumb_resp['content-type'], 'image/jpeg')

# clear thumbnail

response_clear_thumbnail = self.sg.update("Version", self.version['id'], {'image': None})

expected_clear_thumbnail = {'id': self.version['id'], 'image': None, 'type': 'Version'}

self.assertEqual(expected_clear_thumbnail, response_clear_thumbnail)

def test_upload_thumbnail_for_task(self):

"""simple upload thumbnail for task test."""

this_dir, _ = os.path.split(__file__)

path = os.path.abspath(os.path.expanduser(

os.path.join(this_dir, "sg_logo.jpg")))

# upload thumbnail

thumb_id = self.sg.upload_thumbnail("Task", self.task['id'], path)

self.assertTrue(isinstance(thumb_id, int))

# check result on version

task_with_thumbnail = self.sg.find_one('Task', [['id', 'is', self.task['id']]])

task_with_thumbnail = find_one_await_thumbnail(self.sg, 'Task', [['id', 'is', self.task['id']]])

self.assertEqual(task_with_thumbnail.get('type'), 'Task')

self.assertEqual(task_with_thumbnail.get('id'), self.task['id'])

h = Http(".cache")

thumb_resp, content = h.request(task_with_thumbnail.get('image'), "GET")

self.assertEqual(thumb_resp['status'], '200')

self.assertEqual(thumb_resp['content-type'], 'image/jpeg')

# clear thumbnail

response_clear_thumbnail = self.sg.update("Version", self.version['id'], {'image': None})

expected_clear_thumbnail = {'id': self.version['id'], 'image': None, 'type': 'Version'}

self.assertEqual(expected_clear_thumbnail, response_clear_thumbnail)

def test_upload_thumbnail_with_upload_function(self):

"""Upload thumbnail via upload function test"""

path = os.path.abspath(os.path.expanduser(os.path.join(os.path.dirname(__file__), "sg_logo.jpg")))

# upload thumbnail

thumb_id = self.sg.upload("Task", self.task['id'], path, 'image')

self.assertTrue(isinstance(thumb_id, int))

# upload filmstrip thumbnail

f_thumb_id = self.sg.upload("Task", self.task['id'], path, 'filmstrip_image')

self.assertTrue(isinstance(f_thumb_id, int))

def test_requires_direct_s3_upload(self):

"""Test _requires_direct_s3_upload"""

upload_types = self.sg.server_info.get("s3_enabled_upload_types")

direct_uploads_enabled = self.sg.server_info.get("s3_direct_uploads_enabled")

self.sg.server_info["s3_enabled_upload_types"] = None

self.sg.server_info["s3_direct_uploads_enabled"] = None

# Test s3_enabled_upload_types and s3_direct_uploads_enabled not set

self.assertFalse(self.sg._requires_direct_s3_upload("Version", "sg_uploaded_movie"))

self.sg.server_info["s3_enabled_upload_types"] = {

"Version": ["sg_uploaded_movie"]

}

# Test direct_uploads_enabled not set

self.assertFalse(self.sg._requires_direct_s3_upload("Version", "sg_uploaded_movie"))

self.sg.server_info["s3_direct_uploads_enabled"] = True

# Test regular path

self.assertTrue(self.sg._requires_direct_s3_upload("Version", "sg_uploaded_movie"))

self.assertFalse(self.sg._requires_direct_s3_upload("Version", "abc"))

self.assertFalse(self.sg._requires_direct_s3_upload("Abc", "abc"))

# Test star field wildcard and arrays of fields

self.sg.server_info["s3_enabled_upload_types"] = {

"Version": ["sg_uploaded_movie", "test", "other"],

"Test": ["*"],

"Asset": "*"

}

self.assertTrue(self.sg._requires_direct_s3_upload("Version", "sg_uploaded_movie"))

self.assertTrue(self.sg._requires_direct_s3_upload("Version", "test"))

self.assertTrue(self.sg._requires_direct_s3_upload("Version", "other"))

self.assertTrue(self.sg._requires_direct_s3_upload("Test", "abc"))

self.assertTrue(self.sg._requires_direct_s3_upload("Asset", "test"))

# Test default allowed upload type

self.sg.server_info["s3_enabled_upload_types"] = None

self.assertTrue(self.sg._requires_direct_s3_upload("Version", "sg_uploaded_movie"))

self.assertFalse(self.sg._requires_direct_s3_upload("Version", "test"))

# Test star entity_type

self.sg.server_info["s3_enabled_upload_types"] = {

"*": ["sg_uploaded_movie", "test"]

}

self.assertTrue(self.sg._requires_direct_s3_upload("Something", "sg_uploaded_movie"))

self.assertTrue(self.sg._requires_direct_s3_upload("Version", "test"))

self.assertFalse(self.sg._requires_direct_s3_upload("Version", "other"))

# Test entity_type and field_name wildcard

self.sg.server_info["s3_enabled_upload_types"] = {

"*": "*"

}

self.assertTrue(self.sg._requires_direct_s3_upload("Something", "sg_uploaded_movie"))

self.assertTrue(self.sg._requires_direct_s3_upload("Version", "abc"))

self.sg.server_info["s3_enabled_upload_types"] = upload_types

self.sg.server_info["s3_direct_uploads_enabled"] = direct_uploads_enabled

def test_linked_thumbnail_url(self):

this_dir, _ = os.path.split(__file__)

path = os.path.abspath(os.path.expanduser(

os.path.join(this_dir, "sg_logo.jpg")))

thumb_id = self.sg.upload_thumbnail("Project", self.version['project']['id'], path)

response_version_with_project = find_one_await_thumbnail(

self.sg,

"Version",

[["id", "is", self.version["id"]]],

fields=["id", "code", "project.Project.image"],

thumbnail_field_name="project.Project.image"

)

if self.sg.server_caps.version and self.sg.server_caps.version >= (3, 3, 0):

self.assertEqual(response_version_with_project.get('type'), 'Version')

self.assertEqual(response_version_with_project.get('id'), self.version['id'])

self.assertEqual(response_version_with_project.get('code'), self.config.version_code)

h = Http(".cache")

thumb_resp, content = h.request(response_version_with_project.get('project.Project.image'), "GET")

self.assertEqual(thumb_resp['status'], '200')

self.assertEqual(thumb_resp['content-type'], 'image/jpeg')

else:

expected_version_with_project = {

'code': self.config.version_code,

'type': 'Version',

'id': self.version['id'],

'project.Project.image': thumb_id

}

self.assertEqual(expected_version_with_project, response_version_with_project)

# For now skip tests that are erroneously failling on some sites to

# allow CI to pass until the known issue causing this is resolved.

@base.skip("Skipping test that erroneously fails on some sites.")

def test_share_thumbnail(self):

"""share thumbnail between two entities"""

def share_thumbnail_retry(*args, **kwargs):

# Since share_thumbnail raises a ShotgunError if the thumbnail is still

# pending, sleep and retry if this exception is raised, to allow time for

# the thumbnail to finish processing.

thumbnail_id = None

attempts = 0

while attempts < THUMBNAIL_MAX_ATTEMPTS and thumbnail_id is None:

try:

thumbnail_id = self.sg.share_thumbnail(*args, **kwargs)

attempts += 1

except shotgun_api3.ShotgunError:

time.sleep(THUMBNAIL_RETRY_INTERAL)

return thumbnail_id

this_dir, _ = os.path.split(__file__)

path = os.path.abspath(os.path.expanduser(os.path.join(this_dir, "sg_logo.jpg")))

# upload thumbnail to first entity and share it with the rest

share_thumbnail_retry([self.version, self.shot], thumbnail_path=path)

response_version_thumbnail = find_one_await_thumbnail(

self.sg,

'Version',

[['id', 'is', self.version['id']]],

fields=['id', 'code', 'image']

)

response_shot_thumbnail = find_one_await_thumbnail(

self.sg,

'Shot',

[['id', 'is', self.shot['id']]],

fields=['id', 'code', 'image']

)

shot_url = urllib.parse.urlparse(response_shot_thumbnail.get('image'))

version_url = urllib.parse.urlparse(response_version_thumbnail.get('image'))

shot_path = _get_path(shot_url)

version_path = _get_path(version_url)

self.assertEqual(shot_path, version_path)

# share thumbnail from source entity with entities

self.sg.upload_thumbnail("Version", self.version['id'], path)

share_thumbnail_retry([self.asset, self.shot], source_entity=self.version)

response_version_thumbnail = find_one_await_thumbnail(

self.sg,

'Version',

[['id', 'is', self.version['id']]],

fields=['id', 'code', 'image']

)

response_shot_thumbnail = find_one_await_thumbnail(

self.sg,

'Shot',

[['id', 'is', self.shot['id']]],

fields=['id', 'code', 'image']

)

response_asset_thumbnail = find_one_await_thumbnail(

self.sg,

'Asset',

[['id', 'is', self.asset['id']]],

fields=['id', 'code', 'image']

)

shot_url = urllib.parse.urlparse(response_shot_thumbnail.get('image'))

version_url = urllib.parse.urlparse(response_version_thumbnail.get('image'))

asset_url = urllib.parse.urlparse(response_asset_thumbnail.get('image'))

shot_path = _get_path(shot_url)

version_path = _get_path(version_url)

asset_path = _get_path(asset_url)

self.assertEqual(version_path, shot_path)

self.assertEqual(version_path, asset_path)

# raise errors when missing required params or providing conflicting ones

self.assertRaises(shotgun_api3.ShotgunError, self.sg.share_thumbnail,

[self.shot, self.asset], path, self.version)

self.assertRaises(shotgun_api3.ShotgunError, self.sg.share_thumbnail,

[self.shot, self.asset])

@patch('shotgun_api3.Shotgun._send_form')

def test_share_thumbnail_not_ready(self, mock_send_form):

"""throw an exception if trying to share a transient thumbnail"""

mock_send_form.method.assert_called_once()

mock_send_form.return_value = ("2"

"\nsource_entity image is a transient thumbnail that cannot be shared. "

"Try again later, when the final thumbnail is available\n")

self.assertRaises(shotgun_api3.ShotgunThumbnailNotReady, self.sg.share_thumbnail,

[self.version, self.shot], source_entity=self.asset)

@patch('shotgun_api3.Shotgun._send_form')

def test_share_thumbnail_returns_error(self, mock_send_form):

"""throw an exception if server returns an error code"""

mock_send_form.method.assert_called_once()

mock_send_form.return_value = "1\nerror message.\n"

self.assertRaises(shotgun_api3.ShotgunError, self.sg.share_thumbnail,

[self.version, self.shot], source_entity=self.asset)

def test_deprecated_functions(self):

"""Deprecated functions raise errors"""

self.assertRaises(shotgun_api3.ShotgunError, self.sg.schema, "foo")

self.assertRaises(shotgun_api3.ShotgunError, self.sg.entity_types)

def test_simple_summary(self):

"""Test simple call to summarize"""

summaries = [{'field': 'id', 'type': 'count'}]

grouping = [{'direction': 'asc', 'field': 'id', 'type': 'exact'}]

filters = [['project', 'is', self.project]]

result = self.sg.summarize('Shot',

filters=filters,

summary_fields=summaries,

grouping=grouping)

assert(result['groups'])

assert(result['groups'][0]['group_name'])

assert(result['groups'][0]['group_value'])

assert(result['groups'][0]['summaries'])

assert(result['summaries'])

def test_summary_include_archived_projects(self):

"""Test summarize with archived project"""

if self.sg.server_caps.version > (5, 3, 13):

# archive project

self.sg.update('Project', self.project['id'], {'archived': True})

# Ticket #25082 ability to hide archived projects in summary

summaries = [{'field': 'id', 'type': 'count'}]

grouping = [{'direction': 'asc', 'field': 'id', 'type': 'exact'}]

filters = [['project', 'is', self.project]]

result = self.sg.summarize('Shot',

filters=filters,

summary_fields=summaries,

grouping=grouping,

include_archived_projects=False)

self.assertEqual(result['summaries']['id'], 0)

self.sg.update('Project', self.project['id'], {'archived': False})

def test_summary_values(self):

"""Test summarize return data"""

# create three unique shots

shot_prefix = uuid.uuid4().hex

shots = []

shot_data_1 = {

"code": "%s Shot 1" % shot_prefix,

"sg_status_list": "ip",

"sg_cut_duration": 100,

"project": self.project

}

shot_data_2 = {

"code": "%s Shot 2" % shot_prefix,

"sg_status_list": "ip",

"sg_cut_duration": 100,

"project": self.project

}

shot_data_3 = {

"code": "%s Shot 3" % shot_prefix,

"sg_status_list": "fin",

"sg_cut_duration": 100,

"project": self.project

}

shot_data_4 = {

"code": "%s Shot 4" % shot_prefix,

"sg_status_list": "wtg",

"sg_cut_duration": 0,

"project": self.project

}

shots.append(self.sg.create("Shot", shot_data_1))

shots.append(self.sg.create("Shot", shot_data_2))

shots.append(self.sg.create("Shot", shot_data_3))

shots.append(self.sg.create("Shot", shot_data_4))

summaries = [{'field': 'id', 'type': 'count'},

{'field': 'sg_cut_duration', 'type': 'sum'}]

grouping = [{'direction': 'asc',

'field': 'sg_status_list',

'type': 'exact'}]

filters = [['project', 'is', self.project],

['code', 'starts_with', shot_prefix]]

result = self.sg.summarize('Shot',

filters=filters,

summary_fields=summaries,

grouping=grouping)

count = {'id': 4, 'sg_cut_duration': 300}

groups = [

{

'group_name': 'fin',

'group_value': 'fin',

'summaries': {'id': 1, 'sg_cut_duration': 100}

},

{

'group_name': 'ip',

'group_value': 'ip',

'summaries': {'id': 2, 'sg_cut_duration': 200}

},

{

'group_name': 'wtg',

'group_value': 'wtg',

'summaries': {'id': 1, 'sg_cut_duration': 0}

}

]

# clean up

batch_data = []

for s in shots:

batch_data.append({

"request_type": "delete",

"entity_type": "Shot",

"entity_id": s["id"]

})

self.sg.batch(batch_data)

self.assertEqual(result['summaries'], count)

# Do not assume the order of the summarized results.

self.assertEqual(

sorted(

result['groups'],

key=lambda x: x["group_name"]

),

groups

)

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)

result = sg_ascii.find_one('Note', [['id', 'is', self.note['id']]], fields=['content'])

if six.PY2:

# In Python3 there isn't a separate unicode type.

self.assertFalse(_has_unicode(result))

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)

result = sg_unicode.find_one('Note', [['id', 'is', self.note['id']]], fields=['content'])

self.assertTrue(_has_unicode(result))

def test_work_schedule(self):

'''test_work_schedule tests WorkDayRules api'''

self.maxDiff = None

start_date = '2012-01-01'

start_date_obj = datetime.datetime(2012, 1, 1)

end_date = '2012-01-07'

end_date_obj = datetime.datetime(2012, 1, 7)

project = self.project

# We're going to be comparing this value with the value returned from the server, so extract only the type

# and id

user = {"type": self.human_user["type"], "id": self.human_user["id"], "name": self.human_user["name"]}

work_schedule = self.sg.work_schedule_read(start_date, end_date, project, user)

self.assertRaises(shotgun_api3.ShotgunError, self.sg.work_schedule_read,

start_date_obj, end_date_obj, project, user)

resp = self.sg.work_schedule_read(start_date, end_date, project, user)

self.assertEqual(work_schedule, resp)

resp = self.sg.work_schedule_update('2012-01-02', False, 'Studio Holiday')

expected = {

'date': '2012-01-02',

'description': 'Studio Holiday',

'project': None,

'user': None,

'working': False

}

self.assertEqual(expected, resp)

resp = self.sg.work_schedule_read(start_date, end_date, project, user)

work_schedule['2012-01-02'] = {"reason": "STUDIO_EXCEPTION", "working": False, "description": "Studio Holiday"}

self.assertEqual(work_schedule, resp)

resp = self.sg.work_schedule_update('2012-01-03', False, 'Project Holiday', project)

expected = {

'date': '2012-01-03',

'description': 'Project Holiday',

'project': project,

'user': None,

'working': False

}

self.assertEqual(expected, resp)

resp = self.sg.work_schedule_read(start_date, end_date, project, user)

work_schedule['2012-01-03'] = {

"reason": "PROJECT_EXCEPTION",

"working": False,

"description": "Project Holiday"

}

self.assertEqual(work_schedule, resp)

jan4 = datetime.datetime(2012, 1, 4)

self.assertRaises(shotgun_api3.ShotgunError, self.sg.work_schedule_update,

jan4, False, 'Artist Holiday', user=user)

resp = self.sg.work_schedule_update("2012-01-04", False, 'Artist Holiday', user=user)

expected = {

'date': '2012-01-04',

'description': 'Artist Holiday',

'project': None,

'user': user,

'working': False

}

self.assertEqual(expected, resp)

resp = self.sg.work_schedule_read(start_date, end_date, project, user)

work_schedule['2012-01-04'] = {"reason": "USER_EXCEPTION", "working": False, "description": "Artist Holiday"}

# FIXME: There seems to be a regresion on the Shotgun server that needs to be fixed. Disabling the test

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):

return

# This is a big diff if it fails, so show everything.

self.maxDiff = None

# all prefs

resp = self.sg.preferences_read()

expected = {

'date_component_order': 'month_day',

'duration_units': 'days',

'format_currency_fields_decimal_options': '$1,000.99',

'format_currency_fields_display_dollar_sign': False,

'format_currency_fields_negative_options': '- $1,000',

'format_date_fields': '08/04/22 OR 04/08/22 (depending on the Month order preference)',

'format_float_fields': '9,999.99',

'format_float_fields_rounding': '9.999999',

'format_footage_fields': '10-05',

'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

}

# 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

# update the test.

self.assertIn("view_master_settings", resp)

resp.pop("view_master_settings")

self.assertEqual(expected, resp)

# all filtered

resp = self.sg.preferences_read(['date_component_order', 'support_local_storage'])

expected = {

'date_component_order': 'month_day',

'support_local_storage': True

}

self.assertEqual(expected, resp)

# all filtered with invalid pref

resp = self.sg.preferences_read(['date_component_order', 'support_local_storage', 'email_notifications'])

expected = {

'date_component_order': 'month_day',

'support_local_storage': True

}

'''Test fields representing the different data types mapped on the server side.

def setUp(self):

super(TestDataTypes, self).setUp()

def test_set_checkbox(self):

entity = 'HumanUser'

entity_id = self.human_user['id']

field_name = 'email_notes'

pos_values = [False, True]

expected, actual = self.assert_set_field(entity,

entity_id,

field_name,

pos_values)

self.assertEqual(expected, actual)

def test_set_color(self):

entity = 'Task'

entity_id = self.task['id']

field_name = 'color'

pos_values = ['pipeline_step', '222,0,0']

expected, actual = self.assert_set_field(entity,

entity_id,

field_name,

pos_values)

self.assertEqual(expected, actual)

def test_set_date(self):

entity = 'Task'

entity_id = self.task['id']

field_name = 'due_date'

pos_values = ['2008-05-08', '2011-05-05']

expected, actual = self.assert_set_field(entity,

entity_id,

field_name,

pos_values)

self.assertEqual(expected, actual)

def test_set_date_time(self):

entity = 'HumanUser'

entity_id = self.human_user['id']

field_name = 'locked_until'

local = shotgun_api3.shotgun.SG_TIMEZONE.local

dt_1 = datetime.datetime(2008, 10, 13, 23, 10, tzinfo=local)

dt_2 = datetime.datetime(2009, 10, 13, 23, 10, tzinfo=local)

pos_values = [dt_1, dt_2]

expected, actual = self.assert_set_field(entity,

entity_id,

field_name,

pos_values)

self.assertEqual(expected, actual)

def test_set_duration(self):

entity = 'Task'

entity_id = self.task['id']

field_name = 'duration'

pos_values = [2100, 1300]

expected, actual = self.assert_set_field(entity,

entity_id,

field_name,

pos_values)

self.assertEqual(expected, actual)

def test_set_entity(self):

entity = 'Task'

entity_id = self.task['id']

field_name = 'entity'

pos_values = [self.asset, self.shot]

expected, actual = self.assert_set_field(entity,

entity_id,

field_name,

pos_values)

self.assertEqual(expected['id'], actual['id'])

def test_set_float(self):

entity = 'Version'

entity_id = self.version['id']

field_name = 'sg_movie_aspect_ratio'

pos_values = [2.0, 3.0]

expected, actual = self.assert_set_field(entity,

entity_id,

field_name,

pos_values)

self.assertEqual(expected, actual)

def test_set_list(self):

entity = 'Note'

entity_id = self.note['id']