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"

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)

# 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")))

size = os.stat(path).st_size

# test thumbnail upload

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

'project': self.project}

new_version = self.sg.create("Version", data, return_fields=['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('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)

h = Http(".cache")

filmstrip_thumb_resp, content = h.request(new_version.get('filmstrip_image'), "GET")

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

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

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

size = os.stat(path).st_size

# 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']]],

fields=['image'])

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

size = os.stat(path).st_size

# 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']]],

fields=['image'])

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_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 = self.sg.find(

'Version',

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

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

)

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

self.assertEqual(response_version_with_project[0].get('type'), 'Version')

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

self.assertEqual(response_version_with_project[0].get('code'), 'Sg unittest version')

h = Http(".cache")

thumb_resp, content = h.request(response_version_with_project[0].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': 'Sg unittest version',

'type': 'Version',

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

'project.Project.image': thumb_id

}

]

self.assertEqual(expected_version_with_project, response_version_with_project)

def test_share_thumbnail(self):

"""share thumbnail between two entities"""

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

thumbnail_id = self.sg.share_thumbnail(

[self.version, self.shot],

thumbnail_path=path)

response_version_thumbnail = self.sg.find_one(

'Version',

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

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

)

response_shot_thumbnail = self.sg.find_one(

'Shot',

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

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

)

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

version_url = urlparse.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

source_thumbnail_id = self.sg.upload_thumbnail("Version",

self.version['id'], path)

thumbnail_id = self.sg.share_thumbnail(

[self.asset, self.shot],

source_entity=self.version)

response_version_thumbnail = self.sg.find_one(

'Version',

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

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

)

response_shot_thumbnail = self.sg.find_one(

'Shot',

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

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

)

response_asset_thumbnail = self.sg.find_one(

'Asset',

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

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

)

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

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

asset_url = urlparse.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])

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)

self.assertEqual(result['groups'], 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'])

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

user = self.sg.find_one('HumanUser', [['projects', 'is', project]], ['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"}

'''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']

field_name = 'sg_note_type'

pos_values = ['Internal','Client']

expected, actual = self.assert_set_field(entity,

entity_id,

field_name,

pos_values)

self.assertEqual(expected, actual)

def test_set_multi_entity(self):

sg = shotgun_api3.Shotgun( self.config.server_url,

self.config.script_name,

self.config.api_key )

keys = ['project','user','code']

data = {'project':self.project,

'user':self.human_user,

'code':'Alpha'}

version_1 = base._find_or_create_entity(sg, 'Version', data, keys)

data = {'project':self.project,

'user':self.human_user,

'code':'Beta'}

version_2 = base._find_or_create_entity(sg, 'Version', data, keys)

entity = 'Playlist'

entity_id = self.playlist['id']

field_name = 'versions'

# Default set behaviour

pos_values = [[version_1, version_2]]

expected, actual = self.assert_set_field(entity, entity_id,

field_name, pos_values)

self.assertEqual(len(expected), len(actual))

self.assertEqual(

sorted([x['id'] for x in expected]),

sorted([x['id'] for x in actual])

)

# Multi-entity remove mode

pos_values = [[version_1]]

expected, actual = self.assert_set_field(entity, entity_id,

field_name, pos_values, multi_entity_update_mode='remove')

self.assertEqual(1, len(actual))

self.assertEqual(len(expected), len(actual))

self.assertNotEqual(expected[0]['id'],actual[0]['id'])

self.assertEqual(version_2['id'], actual[0]['id'])

# Multi-entity add mode

pos_values = [[version_1]]

expected, actual = self.assert_set_field(entity, entity_id,

field_name, pos_values, multi_entity_update_mode='add')

self.assertEqual(2, len(actual))

self.assertTrue(version_1['id'] in [x['id'] for x in actual])

# Multi-entity set mode

pos_values = [[version_1, version_2]]

expected, actual = self.assert_set_field(entity, entity_id,

field_name, pos_values, multi_entity_update_mode='set')

self.assertEqual(len(expected), len(actual))

self.assertEqual(

sorted([x['id'] for x in expected]),

sorted([x['id'] for x in actual])

)

def test_set_number(self):

entity = 'Shot'

entity_id = self.shot['id']

field_name = 'head_in'

pos_values = [2300, 1300]

expected, actual = self.assert_set_field(entity,

entity_id,

field_name,

pos_values)

self.assertEqual(expected, actual)

def test_set_status_list(self):

entity = 'Task'

entity_id = self.task['id']

field_name = 'sg_status_list'

pos_values = ['rdy','fin']

expected, actual = self.assert_set_field(entity,

entity_id,

field_name,

pos_values)

self.assertEqual(expected, actual)

def test_set_status_list(self):

entity = 'Task'

entity_id = self.task['id']

field_name = 'sg_status_list'

pos_values = ['rdy','fin']

expected, actual = self.assert_set_field(entity,

entity_id,

field_name,

pos_values)

self.assertEqual(expected, actual)

def test_set_tag_list(self):

entity = 'Task'

entity_id = self.task['id']

field_name = 'tag_list'

pos_values = [['a','b'],['c']]

expected, actual = self.assert_set_field(entity,

entity_id,

field_name,

pos_values)

self.assertEqual(expected, actual)

def test_set_text(self):

entity = 'Note'

entity_id = self.note['id']

field_name = 'content'

pos_values = ['this content', 'that content']

expected, actual = self.assert_set_field(entity,

entity_id,

field_name,

pos_values)

self.assertEqual(expected, actual)

def test_set_text_html_entity(self):

entity = 'Note'

entity_id = self.note['id']

field_name = 'content'

pos_values = ['<', '<']

expected, actual = self.assert_set_field(entity,

entity_id,

field_name,

pos_values)

self.assertEqual(expected, actual)

def assert_set_field(self, entity, entity_id, field_name, pos_values, multi_entity_update_mode=None):

query_result = self.sg.find_one(entity,

[['id', 'is', entity_id]],

[field_name])

initial_value = query_result[field_name]

new_value = (initial_value == pos_values[0] and pos_values[1]) or pos_values[0]

if multi_entity_update_mode:

self.sg.update(entity, entity_id, {field_name:new_value},

multi_entity_update_modes={field_name:multi_entity_update_mode})

else:

self.sg.update(entity, entity_id, {field_name:new_value})

new_values = self.sg.find_one(entity,

[['id', 'is', entity_id]],

[field_name])

'''Test utc options'''

def setUp(self):

super(TestUtc, self).setUp()

utc = shotgun_api3.shotgun.SG_TIMEZONE.utc

self.datetime_utc = datetime.datetime(2008, 10, 13, 23, 10, tzinfo=utc)

local = shotgun_api3.shotgun.SG_TIMEZONE.local

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

self.datetime_none = datetime.datetime(2008, 10, 13, 23, 10)

def test_convert_to_utc(self):

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)

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

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)

self._assert_expected(sg_no_utc, self.datetime_none, self.datetime_none)

self._assert_expected(sg_no_utc, self.datetime_utc, self.datetime_none)

def _assert_expected(self, sg, date_time, expected):

entity_name = 'HumanUser'

entity_id = self.human_user['id']

field_name = 'locked_until'

sg.update(entity_name, entity_id, {field_name:date_time})

result = sg.find_one(entity_name, [['id','is',entity_id]],[field_name])

def setUp(self):

super(TestFind, self).setUp()

# We will need the created_at field for the shot

fields = self.shot.keys()[:]

fields.append('created_at')

self.shot = self.sg.find_one('Shot', [['id', 'is', self.shot['id']]], fields)

# We will need the uuid field for our LocalStorage

fields = self.local_storage.keys()[:]

fields.append('uuid')

self.local_storage = self.sg.find_one('LocalStorage', [['id', 'is', self.local_storage['id']]], fields)

def test_find(self):

"""Called find, find_one for known entities"""

filters = []

filters.append(['project','is', self.project])

filters.append(['id','is', self.version['id']])

fields = ['id']

versions = self.sg.find("Version", filters, fields=fields)

self.assertTrue(isinstance(versions, list))

version = versions[0]

self.assertEqual("Version", version["type"])

self.assertEqual(self.version['id'], version["id"])

version = self.sg.find_one("Version", filters, fields=fields)

self.assertEqual("Version", version["type"])

self.assertEqual(self.version['id'], version["id"])

def _id_in_result(self, entity_type, filters, expected_id):

"""

results = self.sg.find(entity_type, filters)

# can't use 'any' in python 2.4

for result in results:

if result['id'] == expected_id:

return True

return False

#TODO test all applicable data types for 'in'

#'currency' => [BigDecimal, Float, NilClass],

#'image' => [Hash, NilClass],

#'percent' => [Bignum, Fixnum, NilClass],

#'serializable' => [Hash, Array, NilClass],

#'system_task_type' => [String, NilClass],

#'timecode' => [Bignum, Fixnum, NilClass],

#'footage' => [Bignum, Fixnum, NilClass, String, Float, BigDecimal],

#'url' => [Hash, NilClass],

#'uuid' => [String],

def test_in_relation_comma_id(self):

"""

filters = [['id', 'in', self.project['id'], 99999]]

result = self._id_in_result('Project', filters, self.project['id'])

self.assertTrue(result)

def test_in_relation_list_id(self):

"""

filters = [['id', 'in', [self.project['id'], 99999]]]

result = self._id_in_result('Project', filters, self.project['id'])

self.assertTrue(result)

def test_not_in_relation_id(self):

"""