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)

# test upload of non-ascii, unicode path

u_path = os.path.abspath(

os.path.expanduser(

glob.glob(os.path.join(unicode(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"

)

# Make sure that non-utf-8 encoded paths raise when they can't be

# converted to utf-8.

#

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

if sys.platform.startswith("win"):

fh = open(file_path_u, "w")

else:

fh = open(file_path_s, "w")

try:

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

finally:

fh.close()

u_path = os.path.abspath(

os.path.expanduser(

glob.glob(os.path.join(unicode(this_dir), u'*.shift-jis'))[0]

)

)

self.assertRaises(

shotgun_api3.ShotgunError,

self.sg.upload,

"Ticket",

self.ticket['id'],

u_path.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'],

u_path,

'attachments',

tag_list="monkeys, everywhere, send, help"

)

# cleanup

os.remove(file_path)

os.remove(u_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)

url = new_version.get('filmstrip_image')

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

self.assertTrue(isinstance(data, str))

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_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 = 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)

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

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

self.assertEqual(work_schedule, resp)

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

# all prefs

resp = self.sg.preferences_read()

expected = {

'date_component_order': 'month_day',

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

'support_local_storage': False

}

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': False

}

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': False

}

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