test_api.py

"""Test calling the Shotgun API functions.

Includes short run tests, like simple crud and single finds. See
test_api_long for other tests.
"""

import datetime
import sys
import os
import re
from mock import patch, Mock, MagicMock
import time
import uuid
import unittest
import urlparse
import urllib2
import warnings
import glob

import shotgun_api3
from shotgun_api3.lib.httplib2 import Http, SSLHandshakeError

import base

class TestShotgunApi(base.LiveTestBase):
    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
        }
        self.assertEqual(expected, resp)


class TestDataTypes(base.LiveTestBase):
    '''Test fields representing the different data types mapped on the server side.

     Untested data types:  password, percent, pivot_column, serializable, image, currency
                           system_task_type, timecode, url, uuid, url_template
    '''
    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)