# -*- coding: utf-8 -*-
#
#      Licensed to the Apache Software Foundation (ASF) under one
#      or more contributor license agreements.  See the NOTICE file
#      distributed with this work for additional information
#      regarding copyright ownership.  The ASF licenses this file
#      to you under the Apache License, Version 2.0 (the
#      "License"); you may not use this file except in compliance
#      with the License.  You may obtain a copy of the License at
#
#        http://www.apache.org/licenses/LICENSE-2.0
#
#      Unless required by applicable law or agreed to in writing,
#      software distributed under the License is distributed on an
#      "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
#      KIND, either express or implied.  See the License for the
#      specific language governing permissions and limitations
#      under the License.
#

"""
Unit tests for cmislib
"""
import unittest
from unittest import TestSuite, TestLoader
from cmislib.model import CmisClient
from cmislib.domain import ACE
from cmislib.exceptions import \
                          ObjectNotFoundException, \
                          CmisException, \
                          NotSupportedException
from cmislib import messages
import os
from time import sleep, time
import settings

## Fix test file paths in case test is launched using nosetests
my_dir = os.path.dirname(os.path.abspath(__file__))
try:
    os.stat(settings.TEST_BINARY_1)
except OSError:
    settings.TEST_BINARY_1 = os.path.join(my_dir, settings.TEST_BINARY_1)
try:
    os.stat(settings.TEST_BINARY_2)
except OSError:
    settings.TEST_BINARY_2 = os.path.join(my_dir, settings.TEST_BINARY_2)


class CmisTestBase(unittest.TestCase):

    """ Common ancestor class for most cmislib unit test classes. """

    def setUp(self):
        """ Create a root test folder for the test. """
        self._cmisClient = CmisClient(settings.REPOSITORY_URL, settings.USERNAME, settings.PASSWORD,
                                      binding=settings.BINDING,
                                      **settings.EXT_ARGS)
        self._repo = self._cmisClient.getDefaultRepository()
        self._rootFolder = self._repo.getObjectByPath(settings.TEST_ROOT_PATH)
        self._folderName = " ".join(['cmislib', self.__class__.__name__, str(time())])
        self._testFolder = self._rootFolder.createFolder(self._folderName)

    def tearDown(self):
        """ Clean up after the test. """
        try:
            self._testFolder.deleteTree()
        except NotSupportedException:
            print "Couldn't delete test folder because deleteTree is not supported"


class CmisClientTest(unittest.TestCase):

    """ Tests for the :class:`CmisClient` class. """

    def testCmisClient(self):
        """Instantiate a CmisClient object"""
        cmisClient = CmisClient(settings.REPOSITORY_URL, settings.USERNAME, settings.PASSWORD,
                                binding=settings.BINDING,
                                **settings.EXT_ARGS)
        self.assert_(cmisClient is not None)

    def testGetRepositories(self):
        """Call getRepositories and make sure at least one comes back with
        an ID and a name
        """
        cmisClient = CmisClient(settings.REPOSITORY_URL, settings.USERNAME, settings.PASSWORD,
                                binding=settings.BINDING,
                                **settings.EXT_ARGS)
        repoInfo = cmisClient.getRepositories()
        self.assert_(len(repoInfo) >= 1)
        self.assert_('repositoryId' in repoInfo[0])
        self.assert_('repositoryName' in repoInfo[0])

    def testDefaultRepository(self):
        """Get the default repository by calling the repo's service URL"""
        cmisClient = CmisClient(settings.REPOSITORY_URL, settings.USERNAME, settings.PASSWORD,
                                binding=settings.BINDING,
                                **settings.EXT_ARGS)
        repo = cmisClient.getDefaultRepository()
        self.assert_(repo is not None)
        self.assert_(repo.getRepositoryId() is not None)

    def testGetRepository(self):
        """Get a repository by repository ID"""
        cmisClient = CmisClient(settings.REPOSITORY_URL, settings.USERNAME, settings.PASSWORD,
                                binding=settings.BINDING,
                                **settings.EXT_ARGS)
        repo = cmisClient.getDefaultRepository()
        defaultRepoId = repo.getRepositoryId()
        defaultRepoName = repo.getRepositoryName()
        repo = cmisClient.getRepository(defaultRepoId)
        self.assertEquals(defaultRepoId, repo.getRepositoryId())
        self.assertEquals(defaultRepoName, repo.getRepositoryName())

    # Error conditions
    def testCmisClientBadUrl(self):
        """Try to instantiate a CmisClient object with a known bad URL"""
        cmisClient = CmisClient(settings.REPOSITORY_URL + 'foobar', settings.USERNAME, settings.PASSWORD,
                                binding=settings.BINDING,
                                **settings.EXT_ARGS)
        self.assertRaises(CmisException, cmisClient.getRepositories)

    def testGetRepositoryBadId(self):
        """Try to get a repository with a bad repo ID"""
        cmisClient = CmisClient(settings.REPOSITORY_URL, settings.USERNAME, settings.PASSWORD,
                                binding=settings.BINDING,
                                **settings.EXT_ARGS)
        self.assertRaises(ObjectNotFoundException,
                          cmisClient.getRepository,
                          '123FOO')


class QueryTest(CmisTestBase):

    """ Tests related to running CMIS queries. """

    # TODO: Test the rest of these queries
    #    queryDateRange = "SELECT cmis:name from cmis:document " \
    #                         "where cmis:creationDate >= TIMESTAMP'2009-11-10T00:00:00.000-06:00' and " \
    #                         "cmis:creationDate < TIMESTAMP'2009-11-18T00:00:00.000-06:00'"
    #    queryFolderFullText = "SELECT cmis:name from cmis:document " \
    #                              "where in_folder('workspace://SpacesStore/3935ce21-9f6f-4d46-9e22-4f97e1d5d9d8') " \
    #                              "and contains('contract')"
    #    queryCombined = "SELECT cmis:name from cmis:document " \
    #                        "where in_tree('workspace://SpacesStore/3935ce21-9f6f-4d46-9e22-4f97e1d5d9d8') and " \
    #                        "contains('contract') and cm:description like \"%sign%\""

    def setUp(self):
        """
        Override the base setUp to include creating a couple
        of test docs.
        """
        CmisTestBase.setUp(self)
        # I think this may be an Alfresco bug. The CMIS query results contain
        # 1 less entry element than the number of search results. So this test
        # will create two documents and search for the second one which should
        # work in all repositories.
        testFile = open(settings.TEST_BINARY_2, 'rb')
        testFileName = settings.TEST_BINARY_2.split('/')[-1]
        self._testContent = self._testFolder.createDocument(testFileName, contentFile=testFile)
        testFile.close()
        testFile = open(settings.TEST_BINARY_2, 'rb')
        self._testContent2 = self._testFolder.createDocument(testFileName.replace('.', '2.'), contentFile=testFile)
        testFile.close()
        self._maxFullTextTries = settings.MAX_FULL_TEXT_TRIES

    def testSimpleSelect(self):
        """Execute simple select star from cmis:document"""
        querySimpleSelect = "SELECT * FROM cmis:document"
        resultSet = self._repo.query(querySimpleSelect)
        self.assertTrue(isInResultSet(resultSet, self._testContent))

    def testWildcardPropertyMatch(self):
        """Find content w/wildcard match on cmis:name property"""
        name = self._testContent.getProperties()['cmis:name']
        querySimpleSelect = "SELECT * FROM cmis:document where cmis:name like '" + name[:7] + "%'"
        resultSet = self._repo.query(querySimpleSelect)
        self.assertTrue(isInResultSet(resultSet, self._testContent))

    def testPropertyMatch(self):
        """Find content matching cmis:name property"""
        name = self._testContent2.getProperties()['cmis:name']
        querySimpleSelect = "SELECT * FROM cmis:document where cmis:name = '" + name + "'"
        resultSet = self._repo.query(querySimpleSelect)
        self.assertTrue(isInResultSet(resultSet, self._testContent2))

    def testFullText(self):
        """Find content using a full-text query"""
        queryFullText = "SELECT cmis:objectId, cmis:name FROM cmis:document " \
                        "WHERE contains('whitepaper')"
        # on the first full text search the indexer may need a chance to
        # do its thing
        found = False
        maxTries = self._maxFullTextTries
        while not found and (maxTries > 0):
            resultSet = self._repo.query(queryFullText)
            found = isInResultSet(resultSet, self._testContent2)
            if not found:
                maxTries -= 1
                print 'Not found...sleeping for 10 secs. Remaining tries:%d' % maxTries
                sleep(settings.FULL_TEXT_WAIT)
        self.assertTrue(found)

    def testScore(self):
        """Find content using FT, sorted by relevance score"""
        queryScore = "SELECT cmis:objectId, cmis:name, Score() as relevance " \
                     "FROM cmis:document WHERE contains('sample') " \
                     "order by relevance DESC"

        # on the first full text search the indexer may need a chance to
        # do its thing
        found = False
        maxTries = self._maxFullTextTries
        while not found and (maxTries > 0):
            resultSet = self._repo.query(queryScore)
            found = isInResultSet(resultSet, self._testContent2)
            if not found:
                maxTries -= 1
                print 'Not found...sleeping for 10 secs. Remaining tries:%d' % maxTries
                sleep(10)
        self.assertTrue(found)


class RepositoryTest(CmisTestBase):

    """ Tests for the :class:`Repository` class. """

    def testRepositoryInfo(self):
        """Retrieve repository info"""
        repoInfo = self._repo.getRepositoryInfo()
        self.assertTrue('repositoryId' in repoInfo)
        self.assertTrue('repositoryName' in repoInfo)
        self.assertTrue('repositoryDescription' in repoInfo)
        self.assertTrue('vendorName' in repoInfo)
        self.assertTrue('productName' in repoInfo)
        self.assertTrue('productVersion' in repoInfo)
        self.assertTrue('rootFolderId' in repoInfo)
        self.assertTrue('cmisVersionSupported' in repoInfo)

    def testRepositoryCapabilities(self):
        """Retrieve repository capabilities"""
        caps = self._repo.getCapabilities()
        self.assertTrue('ACL' in caps)
        self.assertTrue('AllVersionsSearchable' in caps)
        self.assertTrue('Changes' in caps)
        self.assertTrue('ContentStreamUpdatability' in caps)
        self.assertTrue('GetDescendants' in caps)
        self.assertTrue('GetFolderTree' in caps)
        self.assertTrue('Multifiling' in caps)
        self.assertTrue('PWCSearchable' in caps)
        self.assertTrue('PWCUpdatable' in caps)
        self.assertTrue('Query' in caps)
        self.assertTrue('Renditions' in caps)
        self.assertTrue('Unfiling' in caps)
        self.assertTrue('VersionSpecificFiling' in caps)
        self.assertTrue('Join' in caps)

    def testGetRootFolder(self):
        """Get the root folder of the repository"""
        rootFolder = self._repo.getRootFolder()
        self.assert_(rootFolder is not None)
        self.assert_(rootFolder.getObjectId() is not None)

    def testCreateFolder(self):
        """Create a new folder in the root folder"""
        folderName = 'testCreateFolder folder'
        newFolder = self._repo.createFolder(self._rootFolder, folderName)
        self.assertEquals(folderName, newFolder.getName())
        newFolder.delete()

    def testCreateDocument(self):
        """Create a new 'content-less' document"""
        documentName = 'testDocument'
        newDoc = self._repo.createDocument(documentName, parentFolder=self._testFolder)
        self.assertEquals(documentName, newDoc.getName())

    def testCreateDocumentFromString(self):
        """Create a new document from a string"""
        documentName = 'testDocument'
        contentString = 'Test content string'
        newDoc = self._repo.createDocumentFromString(documentName,
                                           parentFolder=self._testFolder,
                                           contentString=contentString,
                                           contentType='text/plain')
        self.assertEquals(documentName, newDoc.getName())
        self.assertEquals(newDoc.getContentStream().read(), contentString)

    # CMIS-279
    def testCreateDocumentUnicode(self):
        """Create a new doc with unicode characters in the name"""
        documentName = u'abc cdeöäüß%§-_caféè.txt'
        newDoc = self._repo.createDocument(documentName, parentFolder=self._testFolder)
        self.assertEquals(documentName, newDoc.getName())

    def testGetObject(self):
        """Create a test folder then attempt to retrieve it as a
        :class:`CmisObject` object using its object ID"""
        folderName = 'testGetObject folder'
        newFolder = self._repo.createFolder(self._testFolder, folderName)
        objectId = newFolder.getObjectId()
        someObject = self._repo.getObject(objectId)
        self.assertEquals(folderName, someObject.getName())
        newFolder.delete()

    def testReturnVersion(self):
        """Get latest and latestmajor versions of an object"""
        f = open(settings.TEST_BINARY_1, 'rb')
        fileName = settings.TEST_BINARY_1.split('/')[-1]
        props = {'cmis:objectTypeId': settings.VERSIONABLE_TYPE_ID}
        doc10 = self._testFolder.createDocument(fileName, contentFile=f, properties=props)
        doc10Id = doc10.getObjectId()
        if not 'canCheckOut' in doc10.allowableActions.keys():
            print 'The test doc cannot be checked out...skipping'
            return
        pwc = doc10.checkout()
        doc11 = pwc.checkin(major='false')  # checkin a minor version, 1.1
        if not 'canCheckOut' in doc11.allowableActions.keys():
            print 'The test doc cannot be checked out...skipping'
            return
        pwc = doc11.checkout()
        doc20 = pwc.checkin()  # checkin a major version, 2.0
        doc20Id = doc20.getObjectId()
        if not 'canCheckOut' in doc20.allowableActions.keys():
            print 'The test doc cannot be checked out...skipping'
            return
        pwc = doc20.checkout()
        doc21 = pwc.checkin(major='false')  # checkin a minor version, 2.1
        doc21Id = doc21.getObjectId()

        docLatest = self._repo.getObject(doc10Id, returnVersion='latest')
        self.assertEquals(doc21Id, docLatest.getObjectId())

        docLatestMajor = self._repo.getObject(doc10Id, returnVersion='latestmajor')
        self.assertEquals(doc20Id, docLatestMajor.getObjectId())

    def testGetFolder(self):
        """Create a test folder then attempt to retrieve the Folder object
        using its object ID"""
        folderName = 'testGetFolder folder'
        newFolder = self._repo.createFolder(self._testFolder, folderName)
        objectId = newFolder.getObjectId()
        someFolder = self._repo.getFolder(objectId)
        self.assertEquals(folderName, someFolder.getName())
        newFolder.delete()

    def testGetObjectByPath(self):
        """Create test objects (one folder, one document) then try to get
        them by path"""
        # names of folders and test docs
        parentFolderName = 'testGetObjectByPath folder'
        subFolderName = 'subfolder'
        docName = 'testdoc'

        # create the folder structure
        parentFolder = self._testFolder.createFolder(parentFolderName)
        subFolder = parentFolder.createFolder(subFolderName)
        # use the subfolder path to get the folder by path
        subFolderPath = subFolder.getProperties().get("cmis:path")
        searchFolder = self._repo.getObjectByPath(subFolderPath)
        self.assertEquals(subFolder.getObjectId(), searchFolder.getObjectId())

        # create a test doc
        doc = subFolder.createDocument(docName)
        # ask the doc for its paths
        searchDocPaths = doc.getPaths()
        # for each path in the list, try to get the object by path
        # this is better than building a path with the doc's name b/c the name
        # isn't guaranteed to be used as the path segment (see CMIS-232)
        for path in searchDocPaths:
            searchDoc = self._repo.getObjectByPath(path)
            self.assertEquals(doc.getObjectId(), searchDoc.getObjectId())

        # get the subfolder by path, then ask for its children
        subFolder = self._repo.getObjectByPath(subFolderPath)
        self.assertEquals(len(subFolder.getChildren().getResults()), 1)

    # getting unfiled documents may work for the atom pub binding for some servers
    # but it isn't part of the spec so removing this test for now
    '''
    def testGetUnfiledDocs(self):
        """Tests the repository's unfiled collection"""

        if not self._repo.getCapabilities()['Unfiling']:
            print 'Repo does not support unfiling, skipping'
            return

        # create a test folder and test doc
        testFolder = self._testFolder.createFolder('unfile test')
        newDoc = testFolder.createDocument('testdoc')

        # make sure the new doc isn't in the unfiled collection
        try:
            rs = self._repo.getUnfiledDocs()
            self.assertFalse(isInResultSet(rs, newDoc))
        except NotSupportedException:
            print 'This repository does not support read access to the unfiled collection...skipping'
            return

        # delete the test folder and tell it to unfile the testdoc
        objId = newDoc.getObjectId()
        testFolder.deleteTree(unfileObjects='unfile')

        # grab the document by object ID
        newDoc = self._repo.getObject(objId)

        # the doc should now be in the unfiled collection
        self.assertTrue(isInResultSet(self._repo.getUnfiledDocs(), newDoc))
        self.assertEquals('testdoc', newDoc.getTitle())
    '''

    # Create document without a parent folder is not yet implemented
    # def testCreateUnfiledDocument(self):
    #     '''Create a new unfiled document'''
    #     if self._repo.getCapabilities()['Unfiling'] != True:
    #         print 'Repo does not support unfiling, skipping'
    #         return
    #     documentName = 'testDocument'
    #     newDoc = self._repo.createDocument(documentName)
    #     self.assertEquals(documentName, newDoc.getName())

    def testMoveDocument(self):
        """Move a Document from one folder to another folder"""
        subFolder1 = self._testFolder.createFolder('sub1')
        doc = subFolder1.createDocument('testdoc1')
        self.assertEquals(len(subFolder1.getChildren()), 1)
        subFolder2 = self._testFolder.createFolder('sub2')
        self.assertEquals(len(subFolder2.getChildren()), 0)
        doc.move(subFolder1, subFolder2)
        self.assertEquals(len(subFolder1.getChildren()), 0)
        self.assertEquals(len(subFolder2.getChildren()), 1)
        self.assertEquals(doc.name, subFolder2.getChildren()[0].name)

    #Exceptions

    def testGetObjectBadId(self):
        """Attempt to get an object using a known bad ID"""
        # this object ID is implementation specific (Alfresco) but is universally
        # bad so it should work for all repositories
        self.assertRaises(ObjectNotFoundException,
                          self._repo.getObject,
                          self._testFolder.getObjectId()[:-5] + 'BADID')

    def testGetObjectBadPath(self):
        """Attempt to get an object using a known bad path"""
        self.assertRaises(ObjectNotFoundException,
                          self._repo.getObjectByPath,
                          '/123foo/BAR.jtp')


class FolderTest(CmisTestBase):

    """ Tests for the :class:`Folder` class """

    def testGetChildren(self):
        """Get the children of the test folder"""
        childFolderName1 = 'testchild1'
        childFolderName2 = 'testchild2'
        grandChildFolderName = 'testgrandchild'
        childFolder1 = self._testFolder.createFolder(childFolderName1)
        childFolder2 = self._testFolder.createFolder(childFolderName2)
        grandChild = childFolder2.createFolder(grandChildFolderName)
        resultSet = self._testFolder.getChildren()
        self.assert_(resultSet is not None)
        self.assertEquals(2, len(resultSet.getResults()))
        self.assertTrue(isInResultSet(resultSet, childFolder1))
        self.assertTrue(isInResultSet(resultSet, childFolder2))
        self.assertFalse(isInResultSet(resultSet, grandChild))

    def testGetDescendants(self):
        """Get the descendants of the root folder"""
        childFolderName1 = 'testchild1'
        childFolderName2 = 'testchild2'
        grandChildFolderName1 = 'testgrandchild'
        childFolder1 = self._testFolder.createFolder(childFolderName1)
        childFolder2 = self._testFolder.createFolder(childFolderName2)
        grandChild = childFolder1.createFolder(grandChildFolderName1)

        # test getting descendants with depth=1
        resultSet = self._testFolder.getDescendants(depth=1)
        self.assert_(resultSet is not None)
        self.assertEquals(2, len(resultSet.getResults()))
        self.assertTrue(isInResultSet(resultSet, childFolder1))
        self.assertTrue(isInResultSet(resultSet, childFolder2))
        self.assertFalse(isInResultSet(resultSet, grandChild))

        # test getting descendants with depth=2
        resultSet = self._testFolder.getDescendants(depth=2)
        self.assert_(resultSet is not None)
        self.assertEquals(3, len(resultSet.getResults()))
        self.assertTrue(isInResultSet(resultSet, childFolder1))
        self.assertTrue(isInResultSet(resultSet, childFolder2))
        self.assertTrue(isInResultSet(resultSet, grandChild))

        # test getting descendants with depth=-1
        resultSet = self._testFolder.getDescendants()  # -1 is the default depth
        self.assert_(resultSet is not None)
        self.assertEquals(3, len(resultSet.getResults()))
        self.assertTrue(isInResultSet(resultSet, childFolder1))
        self.assertTrue(isInResultSet(resultSet, childFolder2))
        self.assertTrue(isInResultSet(resultSet, grandChild))

    def testGetTree(self):
        """Get the folder tree of the test folder"""
        childFolderName1 = 'testchild1'
        childFolderName2 = 'testchild2'
        grandChildFolderName1 = 'testgrandchild'
        childFolder1 = self._testFolder.createFolder(childFolderName1)
        childFolder1.createDocument('testdoc1')
        childFolder2 = self._testFolder.createFolder(childFolderName2)
        childFolder2.createDocument('testdoc2')
        grandChild = childFolder1.createFolder(grandChildFolderName1)
        grandChild.createDocument('testdoc3')

        # test getting tree with depth=1
        resultSet = self._testFolder.getTree(depth=1)
        self.assert_(resultSet is not None)
        self.assertEquals(2, len(resultSet.getResults()))
        self.assertTrue(isInResultSet(resultSet, childFolder1))
        self.assertTrue(isInResultSet(resultSet, childFolder2))
        self.assertFalse(isInResultSet(resultSet, grandChild))

        # test getting tree with depth=2
        resultSet = self._testFolder.getTree(depth=2)
        self.assert_(resultSet is not None)
        self.assertEquals(3, len(resultSet.getResults()))
        self.assertTrue(isInResultSet(resultSet, childFolder1))
        self.assertTrue(isInResultSet(resultSet, childFolder2))
        self.assertTrue(isInResultSet(resultSet, grandChild))

    def testDeleteEmptyFolder(self):
        """Create a test folder, then delete it"""
        folderName = 'testDeleteEmptyFolder folder'
        testFolder = self._testFolder.createFolder(folderName)
        self.assertEquals(folderName, testFolder.getName())
        newFolder = testFolder.createFolder('testFolder')
        testFolderChildren = testFolder.getChildren()
        self.assertEquals(1, len(testFolderChildren.getResults()))
        newFolder.delete()
        testFolderChildren = testFolder.getChildren()
        self.assertEquals(0, len(testFolderChildren.getResults()))

    def testDeleteNonEmptyFolder(self):
        """Create a test folder with something in it, then delete it"""
        folderName = 'testDeleteNonEmptyFolder folder'
        testFolder = self._testFolder.createFolder(folderName)
        self.assertEquals(folderName, testFolder.getName())
        newFolder = testFolder.createFolder('testFolder')
        testFolderChildren = testFolder.getChildren()
        self.assertEquals(1, len(testFolderChildren.getResults()))
        newFolder.createDocument('testDoc')
        self.assertEquals(1, len(newFolder.getChildren().getResults()))
        newFolder.deleteTree()
        testFolderChildren = testFolder.getChildren()
        self.assertEquals(0, len(testFolderChildren.getResults()))

    def testGetProperties(self):
        """Get the root folder, then get its properties"""
        props = self._testFolder.getProperties()
        self.assert_(props is not None)
        self.assert_('cmis:objectId' in props)
        self.assert_(props['cmis:objectId'] is not None)
        self.assert_('cmis:objectTypeId' in props)
        self.assert_(props['cmis:objectTypeId'] is not None)
        self.assert_('cmis:name' in props)
        self.assert_(props['cmis:name'] is not None)

    def testPropertyFilter(self):
        """Test the properties filter"""
        # names of folders and test docs
        parentFolderName = 'testGetObjectByPath folder'
        subFolderName = 'subfolder'

        # create the folder structure
        parentFolder = self._testFolder.createFolder(parentFolderName)
        subFolder = parentFolder.createFolder(subFolderName)
        subFolderPath = subFolder.getProperties().get("cmis:path")

        # Per CMIS-170, CMIS providers are not required to filter the
        # properties returned. So these tests will check only for the presence
        # of the properties asked for, not the absence of properties that
        # should be filtered if the server chooses to do so.

        # test when used with getObjectByPath
        searchFolder = self._repo.getObjectByPath(subFolderPath,
                        filter='cmis:objectId,cmis:objectTypeId,cmis:baseTypeId')
        self.assertEquals(subFolder.getObjectId(), searchFolder.getObjectId())
        self.assertTrue(searchFolder.getProperties().has_key('cmis:objectId'))
        self.assertTrue(searchFolder.getProperties().has_key('cmis:objectTypeId'))
        self.assertTrue(searchFolder.getProperties().has_key('cmis:baseTypeId'))

        # test when used with getObjectByPath + reload
        searchFolder = self._repo.getObjectByPath(subFolderPath,
                        filter='cmis:objectId,cmis:objectTypeId,cmis:baseTypeId')
        searchFolder.reload()
        self.assertEquals(subFolder.getObjectId(), searchFolder.getObjectId())
        self.assertTrue(searchFolder.getProperties().has_key('cmis:objectId'))
        self.assertTrue(searchFolder.getProperties().has_key('cmis:objectTypeId'))
        self.assertTrue(searchFolder.getProperties().has_key('cmis:baseTypeId'))

        # test when used with getObject
        searchFolder = self._repo.getObject(subFolder.getObjectId(),
                        filter='cmis:objectId,cmis:objectTypeId,cmis:baseTypeId')
        self.assertEquals(subFolder.getObjectId(), searchFolder.getObjectId())
        self.assertTrue(searchFolder.getProperties().has_key('cmis:objectId'))
        self.assertTrue(searchFolder.getProperties().has_key('cmis:objectTypeId'))
        self.assertTrue(searchFolder.getProperties().has_key('cmis:baseTypeId'))

        # test when used with getObject + reload
        searchFolder = self._repo.getObject(subFolder.getObjectId(),
                        filter='cmis:objectId,cmis:objectTypeId,cmis:baseTypeId')
        searchFolder.reload()
        self.assertEquals(subFolder.getObjectId(), searchFolder.getObjectId())
        self.assertTrue(searchFolder.getProperties().has_key('cmis:objectId'))
        self.assertTrue(searchFolder.getProperties().has_key('cmis:objectTypeId'))
        self.assertTrue(searchFolder.getProperties().has_key('cmis:baseTypeId'))

        # test that you can do a reload with a reset filter
        searchFolder.reload(filter='*')
        self.assertTrue(searchFolder.getProperties().has_key('cmis:objectId'))
        self.assertTrue(searchFolder.getProperties().has_key('cmis:objectTypeId'))
        self.assertTrue(searchFolder.getProperties().has_key('cmis:baseTypeId'))
        self.assertTrue(searchFolder.getProperties().has_key('cmis:name'))

    def testUpdateProperties(self):
        """Create a test folder, then update its properties"""
        folderName = 'testUpdateProperties folder'
        newFolder = self._testFolder.createFolder(folderName)
        self.assertEquals(folderName, newFolder.getName())
        folderName2 = 'testUpdateProperties folder2'
        props = {'cmis:name': folderName2}
        newFolder.updateProperties(props)
        self.assertEquals(folderName2, newFolder.getName())

    def testSubFolder(self):
        """Create a test folder, then create a test folder within that."""
        parentFolder = self._testFolder.createFolder('testSubFolder folder')
        self.assert_('cmis:objectId' in parentFolder.getProperties())
        childFolder = parentFolder.createFolder('child folder')
        self.assert_('cmis:objectId' in childFolder.getProperties())
        self.assert_(childFolder.getProperties()['cmis:objectId'] is not None)

    def testAllowableActions(self):
        """Create a test folder, then get its allowable actions"""
        actions = self._testFolder.getAllowableActions()
        self.assert_(len(actions) > 0)

    def testGetParent(self):
        """Get a folder's parent using the getParent call"""
        childFolder = self._testFolder.createFolder('parentTest')
        parentFolder = childFolder.getParent()
        self.assertEquals(self._testFolder.getObjectId(), parentFolder.getObjectId())

    def testAddObject(self):
        """Add an existing object to another folder"""
        if not self._repo.getCapabilities()['Multifiling']:
            print 'This repository does not allow multifiling, skipping'
            return

        subFolder1 = self._testFolder.createFolder('sub1')
        doc = subFolder1.createDocument('testdoc1')
        self.assertEquals(len(subFolder1.getChildren()), 1)
        subFolder2 = self._testFolder.createFolder('sub2')
        self.assertEquals(len(subFolder2.getChildren()), 0)
        subFolder2.addObject(doc)
        self.assertEquals(len(subFolder2.getChildren()), 1)
        self.assertEquals(subFolder1.getChildren()[0].name, subFolder2.getChildren()[0].name)

    def testRemoveObject(self):
        """Remove an existing object from a secondary folder"""
        if not self._repo.getCapabilities()['Unfiling']:
            print 'This repository does not allow unfiling, skipping'
            return

        subFolder1 = self._testFolder.createFolder('sub1')
        doc = subFolder1.createDocument('testdoc1')
        self.assertEquals(len(subFolder1.getChildren()), 1)
        subFolder2 = self._testFolder.createFolder('sub2')
        self.assertEquals(len(subFolder2.getChildren()), 0)
        subFolder2.addObject(doc)
        self.assertEquals(len(subFolder2.getChildren()), 1)
        self.assertEquals(subFolder1.getChildren()[0].name, subFolder2.getChildren()[0].name)
        subFolder2.removeObject(doc)
        self.assertEquals(len(subFolder2.getChildren()), 0)
        self.assertEquals(len(subFolder1.getChildren()), 1)
        self.assertEquals(doc.name, subFolder1.getChildren()[0].name)

    def testGetPaths(self):
        """Get a folder's paths"""
        # ask the root for its path
        root = self._repo.getRootFolder()
        paths = root.getPaths()
        self.assertTrue(len(paths) == 1)
        self.assertTrue(paths[0] == '/')
        # ask the test folder for its paths
        paths = self._testFolder.getPaths()
        self.assertTrue(len(paths) == 1)

    # Exceptions

    def testBadParentFolder(self):
        """Try to create a folder on a bad/bogus/deleted parent
        folder object"""
        firstFolder = self._testFolder.createFolder('testBadParentFolder folder')
        self.assert_('cmis:objectId' in firstFolder.getProperties())
        firstFolder.delete()
        # folder isn't in the repo anymore, but I still have the object
        # really, this seems like it ought to be an ObjectNotFoundException but
        # not all CMIS providers report it as such
        self.assertRaises(CmisException,
                          firstFolder.createFolder,
                          'bad parent')

# Per CMIS-169, nothing in the spec says that an exception should be thrown
# when a duplicate folder is created, so this test is really not necessary.
#    def testDuplicateFolder(self):
#        '''Try to create a folder that already exists'''
#        folderName = 'testDupFolder folder'
#        firstFolder = self._testFolder.createFolder(folderName)
#        self.assert_('cmis:objectId' in firstFolder.getProperties())
#        # really, this needs to be ContentAlreadyExistsException but
#        # not all CMIS providers report it as such
#        self.assertRaises(CmisException,
#                          self._testFolder.createFolder,
#                          folderName)


class ChangeEntryTest(CmisTestBase):

    """ Tests for the :class:`ChangeEntry` class """

    def testGetContentChanges(self):

        """Get the content changes and inspect Change Entry props"""

        # need to check changes capability
        changeCap = self._repo.capabilities['Changes']
        if changeCap == 'none':
            print messages.NO_CHANGE_LOG_SUPPORT
            return

        # at least one change should have been made due to the creation of the
        # test documents
        rs = self._repo.getContentChanges()
        self.assertTrue(len(rs) > 0)
        changeEntry = rs[0]
        self.assertTrue(changeEntry.id)
        self.assertTrue(changeEntry.changeType in ['created', 'updated', 'deleted', 'security'])
        self.assertTrue(changeEntry.changeTime)

    def testGetACL(self):

        """Gets the ACL that is included with a Change Entry."""

        # need to check changes capability
        changeCap = self._repo.capabilities['Changes']
        if changeCap == 'none':
            print messages.NO_CHANGE_LOG_SUPPORT
            return

        if changeCap == 'objectidsonly':
            print messages.NO_CHANGE_OBJECT_SUPPORT
            return

        # need to check ACL capability
        if not self._repo.capabilities['ACL']:
            print messages.NO_ACL_SUPPORT
            return

        # need to test once with includeACL set to true
        rs = self._repo.getContentChanges(includeACL='true')
        self.assertTrue(len(rs) > 0)
        changeEntry = rs[0]
        acl = changeEntry.getACL()
        self.assertTrue(acl)
        for entry in acl.getEntries().values():
            self.assertTrue(entry.principalId)
            self.assertTrue(entry.permissions)

        # need to test once without includeACL set
        rs = self._repo.getContentChanges()
        self.assertTrue(len(rs) > 0)
        changeEntry = rs[0]
        acl = changeEntry.getACL()
        self.assertTrue(acl)
        for entry in acl.getEntries().values():
            self.assertTrue(entry.principalId)
            self.assertTrue(entry.permissions)

    def testGetProperties(self):

        """Gets the properties of an object included with a Change Entry."""

        # need to check changes capability
        changeCap = self._repo.capabilities['Changes']
        if changeCap == 'none':
            print messages.NO_CHANGE_LOG_SUPPORT
            return

        if changeCap == 'objectidsonly':
            print messages.NO_CHANGE_OBJECT_SUPPORT
            return

        # need to test once without includeProperties set. the objectID should be there
        rs = self._repo.getContentChanges()
        self.assertTrue(len(rs) > 0)
        changeEntry = rs[0]
        self.assertTrue(changeEntry.properties['cmis:objectId'])

        # need to test once with includeProperties set. the objectID should be there plus object props
        if changeCap in ['properties', 'all']:
            rs = self._repo.getContentChanges(includeProperties='true')
            self.assertTrue(len(rs) > 0)
            changeEntry = rs[0]
            self.assertTrue(changeEntry.properties['cmis:objectId'])
            self.assertTrue(changeEntry.properties['cmis:name'])


class DocumentTest(CmisTestBase):

    """ Tests for the :class:`Document` class """

    def testCheckout(self):
        """Create a document in a test folder, then check it out"""
        props = {'cmis:objectTypeId': settings.VERSIONABLE_TYPE_ID}
        newDoc = self._testFolder.createDocument('testDocument', properties=props)
        if not 'canCheckOut' in newDoc.allowableActions.keys():
            print 'The test doc cannot be checked out...skipping'
            return
        pwcDoc = newDoc.checkout()
        try:
            self.assertTrue(newDoc.isCheckedOut())
            self.assert_('cmis:objectId' in newDoc.getProperties())
            self.assert_('cmis:objectId' in pwcDoc.getProperties())
        finally:
            pwcDoc.delete()

    #CMIS-743
    def testCheckoutAfterFetchByID(self):
        """Create a test doc, fetch it by ID, then check it out"""
        props = {'cmis:objectTypeId': settings.VERSIONABLE_TYPE_ID}
        newDoc = self._testFolder.createDocument('testDocument', properties=props)
        if not 'canCheckOut' in newDoc.allowableActions.keys():
            print 'The test doc cannot be checked out...skipping'
            return
        newDocIdStr = str(newDoc.id)
        newDoc = self._repo.getObject(newDocIdStr)
        pwcDoc = newDoc.checkout()
        try:
            self.assertTrue(newDoc.isCheckedOut())
            self.assert_('cmis:objectId' in newDoc.getProperties())
            self.assert_('cmis:objectId' in pwcDoc.getProperties())
        finally:
            pwcDoc.delete()

    def testCheckin(self):
        """Create a document in a test folder, check it out, then in"""
        testFilename = settings.TEST_BINARY_1.split('/')[-1]
        contentFile = open(testFilename, 'rb')
        props = {'cmis:objectTypeId': settings.VERSIONABLE_TYPE_ID}
        testDoc = self._testFolder.createDocument(testFilename, contentFile=contentFile, properties=props)
        contentFile.close()
        self.assertEquals(testFilename, testDoc.getName())
        if not 'canCheckOut' in testDoc.allowableActions.keys():
            print 'The test doc cannot be checked out...skipping'
            return
        pwcDoc = testDoc.checkout()

        try:
            self.assertTrue(testDoc.isCheckedOut())
            self.assert_('cmis:objectId' in testDoc.getProperties())
            self.assert_('cmis:objectId' in pwcDoc.getProperties())
            testDoc = pwcDoc.checkin()
            self.assertFalse(testDoc.isCheckedOut())
        finally:
            if testDoc.isCheckedOut():
                pwcDoc.delete()

    def testCheckinComment(self):
        """Checkin a document with a comment"""
        testFilename = settings.TEST_BINARY_1.split('/')[-1]
        contentFile = open(testFilename, 'rb')
        props = {'cmis:objectTypeId': settings.VERSIONABLE_TYPE_ID}
        testDoc = self._testFolder.createDocument(testFilename, contentFile=contentFile, properties=props)
        contentFile.close()
        self.assertEquals(testFilename, testDoc.getName())
        if not 'canCheckOut' in testDoc.allowableActions.keys():
            print 'The test doc cannot be checked out...skipping'
            return
        pwcDoc = testDoc.checkout()

        try:
            self.assertTrue(testDoc.isCheckedOut())
            testDoc = pwcDoc.checkin(checkinComment='Just a few changes')
            self.assertFalse(testDoc.isCheckedOut())
            self.assertEquals('Just a few changes',
                          testDoc.getProperties()['cmis:checkinComment'])
        finally:
            if testDoc.isCheckedOut():
                pwcDoc.delete()

    def testCheckinAfterGetPWC(self):
        """Create a document in a test folder, check it out, call getPWC, then checkin"""
        if not self._repo.getCapabilities()['PWCUpdatable'] == True:
            print 'Repository does not support PWCUpdatable, skipping'
            return

        testFilename = settings.TEST_BINARY_1.split('/')[-1]
        contentFile = open(testFilename, 'rb')
        props = {'cmis:objectTypeId': settings.VERSIONABLE_TYPE_ID}
        testDoc = self._testFolder.createDocument(testFilename, contentFile=contentFile, properties=props)
        contentFile.close()
        self.assertEquals(testFilename, testDoc.getName())
        # Alfresco has a bug where if you get the PWC this way
        # the checkin will not be successful
        if not 'canCheckOut' in testDoc.allowableActions.keys():
            print 'The test doc cannot be checked out...skipping'
            return
        testDoc.checkout()
        pwcDoc = testDoc.getPrivateWorkingCopy()
        try:
            self.assertTrue(testDoc.isCheckedOut())
            self.assert_('cmis:objectId' in testDoc.getProperties())
            self.assert_('cmis:objectId' in pwcDoc.getProperties())
            testDoc = pwcDoc.checkin()
            self.assertFalse(testDoc.isCheckedOut())
        finally:
            if testDoc.isCheckedOut():
                pwcDoc.delete()

    def testCancelCheckout(self):
        """Create a document in a test folder, check it out, then cancel
        checkout"""
        props = {'cmis:objectTypeId': settings.VERSIONABLE_TYPE_ID}
        newDoc = self._testFolder.createDocument('testDocument', properties=props)
        if not 'canCheckOut' in newDoc.allowableActions.keys():
            print 'The test doc cannot be checked out...skipping'
            return
        pwcDoc = newDoc.checkout()
        try:
            self.assertTrue(newDoc.isCheckedOut())
            self.assert_('cmis:objectId' in newDoc.getProperties())
            self.assert_('cmis:objectId' in pwcDoc.getProperties())
        finally:
            pwcDoc.delete()
        self.assertFalse(newDoc.isCheckedOut())

    def testDeleteDocument(self):
        """Create a document in a test folder, then delete it"""
        newDoc = self._testFolder.createDocument('testDocument')
        children = self._testFolder.getChildren()
        self.assertEquals(1, len(children.getResults()))
        newDoc.delete()
        children = self._testFolder.getChildren()
        self.assertEquals(0, len(children.getResults()))

    def testGetLatestVersion(self):
        """Get latest version of an object"""
        f = open(settings.TEST_BINARY_1, 'rb')
        fileName = settings.TEST_BINARY_1.split('/')[-1]
        props = {'cmis:objectTypeId': settings.VERSIONABLE_TYPE_ID}
        doc10 = self._testFolder.createDocument(fileName, contentFile=f, properties=props)
        if not 'canCheckOut' in doc10.allowableActions.keys():
            print 'The test doc cannot be checked out...skipping'
            return
        pwc = doc10.checkout()
        doc11 = pwc.checkin(major='false')  # checkin a minor version, 1.1
        if not 'canCheckOut' in doc11.allowableActions.keys():
            print 'The test doc cannot be checked out...skipping'
            return
        pwc = doc11.checkout()
        doc20 = pwc.checkin()  # checkin a major version, 2.0
        doc20Id = doc20.getObjectId()
        if not 'canCheckOut' in doc20.allowableActions.keys():
            print 'The test doc cannot be checked out...skipping'
            return
        pwc = doc20.checkout()
        doc21 = pwc.checkin(major='false')  # checkin a minor version, 2.1
        doc21Id = doc21.getObjectId()

        docLatest = doc10.getLatestVersion()
        self.assertEquals(doc21Id, docLatest.getObjectId())

        docLatestMajor = doc10.getLatestVersion(major='true')
        self.assertEquals(doc20Id, docLatestMajor.getObjectId())

    def testGetPropertiesOfLatestVersion(self):
        """Get properties of latest version of an object"""
        f = open(settings.TEST_BINARY_1, 'rb')
        fileName = settings.TEST_BINARY_1.split('/')[-1]
        props = {'cmis:objectTypeId': settings.VERSIONABLE_TYPE_ID}
        doc10 = self._testFolder.createDocument(fileName, contentFile=f, properties=props)
        if not 'canCheckOut' in doc10.allowableActions.keys():
            print 'The test doc cannot be checked out...skipping'
            return
        pwc = doc10.checkout()
        doc11 = pwc.checkin(major='false')  # checkin a minor version, 1.1
        if not 'canCheckOut' in doc11.allowableActions.keys():
            print 'The test doc cannot be checked out...skipping'
            return
        pwc = doc11.checkout()
        doc20 = pwc.checkin()  # checkin a major version, 2.0
        # what comes back from a checkin may not include all props, so reload
        doc20.reload()
        doc20Label = doc20.getProperties()['cmis:versionLabel']
        if not 'canCheckOut' in doc20.allowableActions.keys():
            print 'The test doc cannot be checked out...skipping'
            return
        pwc = doc20.checkout()
        doc21 = pwc.checkin(major='false')  # checkin a minor version, 2.1
        # what comes back from a checkin may not include all props, so reload
        doc21.reload()
        doc21Label = doc21.getProperties()['cmis:versionLabel']

        propsLatest = doc10.getPropertiesOfLatestVersion()
        self.assertEquals(doc21Label, propsLatest['cmis:versionLabel'])

        propsLatestMajor = doc10.getPropertiesOfLatestVersion(major='true')
        self.assertEquals(doc20Label, propsLatestMajor['cmis:versionLabel'])

    def testGetProperties(self):
        """Create a document in a test folder, then get its properties"""
        newDoc = self._testFolder.createDocument('testDocument')
        self.assertEquals('testDocument', newDoc.getName())
        self.assertTrue('cmis:objectTypeId' in newDoc.getProperties())
        self.assertTrue('cmis:objectId' in newDoc.getProperties())

    def testAllowableActions(self):
        """Create document in a test folder, then get its allowable actions"""
        newDoc = self._testFolder.createDocument('testDocument')
        actions = newDoc.getAllowableActions()
        self.assert_(len(actions) > 0)

    def testUpdateProperties(self):
        """Create a document in a test folder, then update its properties"""
        newDoc = self._testFolder.createDocument('testDocument')
        self.assertEquals('testDocument', newDoc.getName())
        props = {'cmis:name': 'testDocument2'}
        newDoc.updateProperties(props)
        self.assertEquals('testDocument2', newDoc.getName())

    def testSetContentStreamPWC(self):
        """Set the content stream on the PWC"""
        if self._repo.getCapabilities()['ContentStreamUpdatability'] == 'none':
            print 'This repository does not allow content stream updates, skipping'
            return

        testFile1 = settings.TEST_BINARY_1
        testFile1Size = os.path.getsize(testFile1)
        exportFile1 = testFile1.replace('.', 'export.')
        testFile2 = settings.TEST_BINARY_2
        testFile2Size = os.path.getsize(testFile2)
        exportFile2 = testFile1.replace('.', 'export.')

        # create a test document
        contentFile = open(testFile1, 'rb')
        newDoc = self._testFolder.createDocument(testFile1, contentFile=contentFile)
        contentFile.close()

        # export the test document
        result = newDoc.getContentStream()
        outfile = open(exportFile1, 'wb')
        outfile.write(result.read())
        result.close()
        outfile.close()

        # the file we exported should be the same size as the file we
        # originally created
        self.assertEquals(testFile1Size, os.path.getsize(exportFile1))

        # checkout the file
        if newDoc.allowableActions.has_key('canCheckOut') and \
                newDoc.allowableActions['canCheckOut'] == True:
            pass
        else:
            print 'The test doc cannot be checked out...skipping'
            return
        pwc = newDoc.checkout()

        # update the PWC with a new file
        f = open(testFile2, 'rb')
        pwc.setContentStream(f)
        f.close()

        # checkin the PWC
        newDoc = pwc.checkin()

        # export the checked in document
        result = newDoc.getContentStream()
        outfile = open(exportFile2, 'wb')
        outfile.write(result.read())
        result.close()
        outfile.close()

        # the file we exported should be the same size as the file we
        # checked in after updating the PWC
        self.assertEquals(testFile2Size, os.path.getsize(exportFile2))
        os.remove(exportFile2)

    def testSetContentStreamPWCMimeType(self):
        """Check the mimetype after the PWC checkin"""
        if self._repo.getCapabilities()['ContentStreamUpdatability'] == 'none':
            print 'This repository does not allow content stream updates, skipping'
            return

        testFile1 = settings.TEST_BINARY_1
        fileName = testFile1.split('/')[-1]

        # create a test document
        contentFile = open(testFile1, 'rb')
        props = {'cmis:objectTypeId': settings.VERSIONABLE_TYPE_ID}
        newDoc = self._testFolder.createDocument(fileName, contentFile=contentFile, properties=props)
        origMimeType = newDoc.properties['cmis:contentStreamMimeType']
        contentFile.close()

        # checkout the file
        if not 'canCheckOut' in newDoc.allowableActions.keys():
            print 'The test doc cannot be checked out...skipping'
            return
        pwc = newDoc.checkout()

        # update the PWC with a new file
        f = open(testFile1, 'rb')
        pwc.setContentStream(f)
        f.close()

        # checkin the PWC
        newDoc = pwc.checkin()

        # CMIS-231 the checked in doc should have the same mime type as
        # the original document
        self.assertEquals(origMimeType,
                          newDoc.properties['cmis:contentStreamMimeType'])

    def testSetContentStreamDoc(self):
        """Set the content stream on a doc that's not checked out"""
        if self._repo.getCapabilities()['ContentStreamUpdatability'] != 'anytime':
            print 'This repository does not allow content stream updates on the doc, skipping'
            return

        testFile1 = settings.TEST_BINARY_1
        testFile1Size = os.path.getsize(testFile1)
        exportFile1 = testFile1.replace('.', 'export.')
        testFile2 = settings.TEST_BINARY_2
        testFile2Size = os.path.getsize(testFile2)
        exportFile2 = testFile1.replace('.', 'export.')

        # create a test document
        contentFile = open(testFile1, 'rb')
        fileName = testFile1.split('/')[-1]
        newDoc = self._testFolder.createDocument(fileName, contentFile=contentFile)
        contentFile.close()

        # export the test document
        result = newDoc.getContentStream()
        outfile = open(exportFile1, 'wb')
        outfile.write(result.read())
        result.close()
        outfile.close()

        # the file we exported should be the same size as the file we
        # originally created
        self.assertEquals(testFile1Size, os.path.getsize(exportFile1))

        # update the PWC with a new file
        f = open(testFile2, 'rb')
        newDoc.setContentStream(f)
        f.close()

        # export the checked in document
        result = newDoc.getContentStream()
        outfile = open(exportFile2, 'wb')
        outfile.write(result.read())
        result.close()
        outfile.close()

        # the file we exported should be the same size as the file we
        # checked in after updating the PWC
        self.assertEquals(testFile2Size, os.path.getsize(exportFile2))
        os.remove(exportFile2)

    def testDeleteContentStreamPWC(self):
        """Delete the content stream of a PWC"""
        if self._repo.getCapabilities()['ContentStreamUpdatability'] == 'none':
            print 'This repository does not allow content stream updates, skipping'
            return
        if not self._repo.getCapabilities()['PWCUpdatable'] == True:
            print 'Repository does not support PWCUpdatable, skipping'
            return

        # create a test document
        contentFile = open(settings.TEST_BINARY_1, 'rb')
        props = {'cmis:objectTypeId': settings.VERSIONABLE_TYPE_ID}
        fileName = settings.TEST_BINARY_1.split('/')[-1]
        newDoc = self._testFolder.createDocument(fileName, contentFile=contentFile, properties=props)
        contentFile.close()
        if not 'canCheckOut' in newDoc.allowableActions.keys():
            print 'The test doc cannot be checked out...skipping'
            return
        pwc = newDoc.checkout()
        pwc.deleteContentStream()
        self.assertRaises(CmisException, pwc.getContentStream)
        pwc.delete()

    def testCreateDocumentBinary(self):
        """Create a binary document using a file from the file system"""
        testFilename = settings.TEST_BINARY_1
        contentFile = open(testFilename, 'rb')
        newDoc = self._testFolder.createDocument(testFilename, contentFile=contentFile)
        contentFile.close()
        self.assertEquals(testFilename, newDoc.getName())

        # test to make sure the file we get back is the same length
        # as the file we sent
        result = newDoc.getContentStream()
        exportFilename = testFilename.replace('.', 'export.')
        outfile = open(exportFilename, 'wb')
        outfile.write(result.read())
        result.close()
        outfile.close()
        self.assertEquals(os.path.getsize(testFilename),
                          os.path.getsize(exportFilename))

        # cleanup
        os.remove(exportFilename)

    def testCreateDocumentFromString(self):
        """Create a new document from a string"""
        documentName = 'testDocument'
        contentString = 'Test content string'
        newDoc = self._testFolder.createDocumentFromString(documentName,
            contentString=contentString, contentType='text/plain')
        self.assertEquals(documentName, newDoc.getName())
        self.assertEquals(newDoc.getContentStream().read(), contentString)

    def testCreateDocumentPlain(self):
        """Create a plain document using a file from the file system"""
        testFilename = 'plain.txt'
        testFile = open(testFilename, 'w')
        testFile.write('This is a sample text file line 1.\n')
        testFile.write('This is a sample text file line 2.\n')
        testFile.write('This is a sample text file line 3.\n')
        testFile.close()
        contentFile = open(testFilename, 'r')
        newDoc = self._testFolder.createDocument(testFilename, contentFile=contentFile)
        contentFile.close()
        self.assertEquals(testFilename, newDoc.getName())

        # test to make sure the file we get back is the same length as the
        # file we sent
        result = newDoc.getContentStream()
        exportFilename = testFilename.replace('txt', 'export.txt')
        outfile = open(exportFilename, 'w')
        outfile.write(result.read())
        result.close()
        outfile.close()
        self.assertEquals(os.path.getsize(testFilename),
                          os.path.getsize(exportFilename))

        # export
        os.remove(exportFilename)
        os.remove(testFilename)

    def testGetAllVersions(self):
        """Get all versions of an object"""
        props = {'cmis:objectTypeId': settings.VERSIONABLE_TYPE_ID}
        testDoc = self._testFolder.createDocument('testdoc', properties=props)
        if not 'canCheckOut' in testDoc.allowableActions.keys():
            print 'The test doc cannot be checked out...skipping'
            return
        pwc = testDoc.checkout()
        doc = pwc.checkin()  # 2.0
        if not 'canCheckOut' in doc.allowableActions.keys():
            print 'The test doc cannot be checked out...skipping'
            return
        pwc = doc.checkout()
        doc = pwc.checkin()  # 3.0
        # what comes back from a checkin may not include all props, so reload
        doc.reload()
        # InMemory 0.9 is using 'V 3.0' so this test fails with that server
        #self.assertEquals('3.0', doc.getProperties()['cmis:versionLabel'])
        rs = doc.getAllVersions()
        self.assertEquals(3, len(rs.getResults()))
#        for count in range(0, 3):
#            if count == 0:
#                self.assertEquals('true',
#                             rs.getResults().values()[count].getProperties()['cmis:isLatestVersion'])
#            else:
#                self.assertEquals('false',
#                             rs.getResults().values()[count].getProperties()['cmis:isLatestVersion'])

    def testGetObjectParents(self):
        """Gets all object parents of an CmisObject"""
        childFolder = self._testFolder.createFolder('parentTest')
        parentFolder = childFolder.getObjectParents().getResults()[0]
        self.assertEquals(self._testFolder.getObjectId(), parentFolder.getObjectId())

    def testGetObjectParentsWithinRootFolder(self):
        """Gets all object parents of a root folder"""
        rootFolder = self._repo.getRootFolder()
        self.assertRaises(NotSupportedException, rootFolder.getObjectParents)

    def testGetObjectParentsMultiple(self):
        """Gets all parents of a multi-filed object"""
        if not self._repo.getCapabilities()['Multifiling']:
            print 'This repository does not allow multifiling, skipping'
            return

        subFolder1 = self._testFolder.createFolder('sub1')
        doc = subFolder1.createDocument('testdoc1')
        self.assertEquals(len(subFolder1.getChildren()), 1)
        subFolder2 = self._testFolder.createFolder('sub2')
        self.assertEquals(len(subFolder2.getChildren()), 0)
        subFolder2.addObject(doc)
        self.assertEquals(len(subFolder2.getChildren()), 1)
        self.assertEquals(subFolder1.getChildren()[0].name, subFolder2.getChildren()[0].name)
        parentNames = ['sub1', 'sub2']
        for parent in doc.getObjectParents():
            parentNames.remove(parent.name)
        self.assertEquals(len(parentNames), 0)

    def testGetPaths(self):
        """Get the paths of a document"""
        testDoc = self._testFolder.createDocument('testdoc')
        # ask the test doc for its paths
        paths = testDoc.getPaths()
        self.assertTrue(len(paths) >= 1)

    def testRenditions(self):
        """Get the renditions for a document"""
        if not self._repo.getCapabilities().has_key('Renditions'):
            print 'Repo does not support unfiling, skipping'
            return

        testDoc = self._testFolder.createDocumentFromString('testdoc.txt', contentString='test', contentType='text/plain')
        sleep(settings.FULL_TEXT_WAIT)
        if (testDoc.getAllowableActions().has_key('canGetRenditions') and
            testDoc.getAllowableActions()['canGetRenditions'] == True):
            rends = testDoc.getRenditions()
            self.assertTrue(len(rends) >= 1)
        else:
            print 'Test doc does not have rendition, skipping'
            return


class TypeTest(unittest.TestCase):

    """
    Tests for the :class:`ObjectType` class (and related methods in the
    :class:`Repository` class.
    """

    def testTypeDescendants(self):
        """Get the descendant types of the repository."""

        cmisClient = CmisClient(settings.REPOSITORY_URL, settings.USERNAME, settings.PASSWORD,
                                binding=settings.BINDING,
                                **settings.EXT_ARGS)
        repo = cmisClient.getDefaultRepository()
        typeDefs = repo.getTypeDescendants()
        folderDef = None
        for typeDef in typeDefs:
            if typeDef.getTypeId() == 'cmis:folder':
                folderDef = typeDef
                break
        self.assertTrue(folderDef)
        self.assertTrue(folderDef.baseId)

    def testTypeChildren(self):
        """Get the child types for this repository and make sure cmis:folder
        is in the list."""

        #This test would be more interesting if there was a standard way to
        #deploy a custom model. Then we could look for custom types.

        cmisClient = CmisClient(settings.REPOSITORY_URL, settings.USERNAME, settings.PASSWORD,
                                binding=settings.BINDING,
                                **settings.EXT_ARGS)
        repo = cmisClient.getDefaultRepository()
        typeDefs = repo.getTypeChildren()
        folderDef = None
        for typeDef in typeDefs:
            if typeDef.getTypeId() == 'cmis:folder':
                folderDef = typeDef
                break
        self.assertTrue(folderDef)
        self.assertTrue(folderDef.baseId)

    def testTypeDefinition(self):
        """Get the cmis:document type and test a few props of the type."""
        cmisClient = CmisClient(settings.REPOSITORY_URL, settings.USERNAME, settings.PASSWORD,
                                binding=settings.BINDING,
                                **settings.EXT_ARGS)
        repo = cmisClient.getDefaultRepository()
        docTypeDef = repo.getTypeDefinition('cmis:document')
        self.assertEquals('cmis:document', docTypeDef.getTypeId())
        self.assertTrue(docTypeDef.baseId)

    def testTypeProperties(self):
        """Get the properties for a type."""
        cmisClient = CmisClient(settings.REPOSITORY_URL, settings.USERNAME, settings.PASSWORD,
                                binding=settings.BINDING,
                                **settings.EXT_ARGS)
        repo = cmisClient.getDefaultRepository()
        docTypeDef = repo.getTypeDefinition('cmis:document')
        self.assertEquals('cmis:document', docTypeDef.getTypeId())
        props = docTypeDef.getProperties().values()
        self.assertTrue(len(props) > 0)
        for prop in props:
            if prop.queryable:
                self.assertTrue(prop.queryName)
            self.assertTrue(prop.propertyType)


class ACLTest(CmisTestBase):

    """
    Tests related to :class:`ACL` and :class:`ACE`
    """

    def testSupportedPermissions(self):
        """Test the value of supported permissions enum"""
        if not self._repo.getCapabilities()['ACL']:
            print messages.NO_ACL_SUPPORT
            return
        self.assertTrue(self._repo.getSupportedPermissions() in ['basic', 'repository', 'both'])

    def testPermissionDefinitions(self):
        """Test the list of permission definitions"""
        if not self._repo.getCapabilities()['ACL']:
            print messages.NO_ACL_SUPPORT
            return
        supportedPerms = self._repo.getPermissionDefinitions()
        self.assertTrue(supportedPerms.has_key('cmis:write'))

    def testPermissionMap(self):
        """Test the permission mapping"""
        if not self._repo.getCapabilities()['ACL']:
            print messages.NO_ACL_SUPPORT
            return
        permMap = self._repo.getPermissionMap()
        self.assertTrue(permMap.has_key('canGetProperties.Object'))
        self.assertTrue(len(permMap['canGetProperties.Object']) > 0)

    def testPropagation(self):
        """Test the propagation setting"""
        if not self._repo.getCapabilities()['ACL']:
            print messages.NO_ACL_SUPPORT
            return
        self.assertTrue(self._repo.getPropagation() in ['objectonly', 'propagate', 'repositorydetermined'])

    def testGetObjectACL(self):
        """Test getting an object's ACL"""
        if not self._repo.getCapabilities()['ACL']:
            print messages.NO_ACL_SUPPORT
            return
        acl = self._testFolder.getACL()
        for entry in acl.getEntries().values():
            self.assertTrue(entry.principalId)
            self.assertTrue(entry.permissions)

    def testApplyACL(self):
        """Test updating an object's ACL"""
        if not self._repo.getCapabilities()['ACL']:
            print messages.NO_ACL_SUPPORT
            return
        if not self._repo.getCapabilities()['ACL'] == 'manage':
            print 'Repository does not support manage ACL'
            return
        if not self._repo.getSupportedPermissions() in ['both', 'basic']:
            print 'Repository needs to support either both or basic permissions for this test'
            return
        acl = self._testFolder.getACL()
        acl.addEntry(settings.TEST_PRINCIPAL_ID, 'cmis:write')
        acl = self._testFolder.applyACL(acl)
        # would be good to check that the permission we get back is what we set
        # but at least one server (Alf) appears to map the basic perm to a
        # repository-specific perm
        self.assertTrue(acl.getEntries().has_key(settings.TEST_PRINCIPAL_ID))


def isInCollection(collection, targetDoc):
    """
    Util function that searches a list of objects for a matching target
    object.
    """
    for doc in collection:
        # hacking around a bizarre thing in Alfresco which is that when the
        # PWC comes back it has an object ID of say 123ABC but when you look
        # in the checked out collection the object ID of the PWC is now
        # 123ABC;1.0. What is that ;1.0? I don't know, but object IDs are
        # supposed to be immutable so I'm not sure what's going on there.
        if doc.getObjectId().startswith(targetDoc.getObjectId()):
            return True
    return False


def isInResultSet(resultSet, targetDoc):
    """
    Util function that searches a :class:`ResultSet` for a specified target
    object. Note that this function will do a getNext on every page of the
    result set until it finds what it is looking for or reaches the end of
    the result set. For every item in the result set, the properties
    are retrieved. Long story short: this could be an expensive call.
    """
    done = False
    while not done:
        if resultSet.hasObject(targetDoc.getObjectId()):
            return True
        if resultSet.hasNext():
            resultSet.getNext()
        else:
            done = True

if __name__ == "__main__":
    #unittest.main()
    tts = TestSuite()
    #tts.addTests(TestLoader().loadTestsFromName('testGetObjectByPath', RepositoryTest))
    #unittest.TextTestRunner().run(tts)
    #import sys; sys.exit(0)

    tts.addTests(TestLoader().loadTestsFromTestCase(CmisClientTest))
    tts.addTests(TestLoader().loadTestsFromTestCase(RepositoryTest))
    tts.addTests(TestLoader().loadTestsFromTestCase(FolderTest))
    tts.addTests(TestLoader().loadTestsFromTestCase(DocumentTest))
    tts.addTests(TestLoader().loadTestsFromTestCase(TypeTest))
    tts.addTests(TestLoader().loadTestsFromTestCase(ACLTest))
    tts.addTests(TestLoader().loadTestsFromTestCase(ChangeEntryTest))

#    tts.addTests(TestLoader().loadTestsFromName('testCreateDocumentFromString', RepositoryTest))
#    tts.addTests(TestLoader().loadTestsFromName('testCreateDocumentFromString', DocumentTest))
#    tts.addTests(TestLoader().loadTestsFromName('testMoveDocument', RepositoryTest))
#    tts.addTests(TestLoader().loadTestsFromName('testCreateDocumentBinary', DocumentTest))
#    tts.addTests(TestLoader().loadTestsFromName('testCreateDocumentPlain', DocumentTest))
#    tts.addTests(TestLoader().loadTestsFromName('testAddObject', FolderTest))
#    tts.addTests(TestLoader().loadTestsFromName('testRemoveObject', FolderTest))
#    tts.addTests(TestLoader().loadTestsFromName('testFolderLeadingDot', FolderTest))
#    tts.addTests(TestLoader().loadTestsFromName('testGetObjectParents', DocumentTest))
#    tts.addTests(TestLoader().loadTestsFromName('testGetObjectParentsMultiple', DocumentTest))
#    tts.addTests(TestLoader().loadTestsFromName('testRenditions', DocumentTest))

    # WARNING: Potentially long-running tests

    # Query tests
    #tts.addTests(TestLoader().loadTestsFromTestCase(QueryTest))
    #tts.addTest(QueryTest('testPropertyMatch'))
    #tts.addTest(QueryTest('testFullText'))
    #tts.addTest(QueryTest('testScore'))
    #tts.addTest(QueryTest('testWildcardPropertyMatch'))
    #tts.addTest(QueryTest('testSimpleSelect'))

    unittest.TextTestRunner().run(tts)