You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@chemistry.apache.org by jp...@apache.org on 2010/02/20 21:33:00 UTC
svn commit: r912218 [11/11] - in /incubator/chemistry/trunk/cmislib: ./ src/
src/cmislib.egg-info/ src/cmislib/ src/data/ src/doc/ src/doc/src/
src/doc/src/.doctrees/ src/doc/src/_static/ src/tests/
Added: incubator/chemistry/trunk/cmislib/src/doc/src/examples.rst
URL: http://svn.apache.org/viewvc/incubator/chemistry/trunk/cmislib/src/doc/src/examples.rst?rev=912218&view=auto
==============================================================================
--- incubator/chemistry/trunk/cmislib/src/doc/src/examples.rst (added)
+++ incubator/chemistry/trunk/cmislib/src/doc/src/examples.rst Sat Feb 20 20:32:57 2010
@@ -0,0 +1,119 @@
+.. _examples:
+
+========
+Examples
+========
+There's nothing in cmislib that is specific to any particular vendor. Once you give it your CMIS provider's service URL and some credentials, it figures out where to go from there. But I haven't tested with anything other than Alfresco yet, and this thing is still hot out of the oven. If you want to help test it against other CMIS 1.0cd04 repositories I'd love the help.
+
+Anyway, let's look at some examples using Alfresco's public CMIS repository.
+
+-----------------------
+Get a Repository object
+-----------------------
+
+ #. From the command-line, start the Python shell by typing `python` then hit enter.
+ #. Import the CmisClient:
+
+ >>> from cmislib.model import CmisClient
+
+ #. Point the CmisClient at the repository's service URL
+
+ >>> client = CmisClient('http://cmis.alfresco.com/s/cmis', 'admin', 'admin')
+
+ #. Get the default repository for the service
+
+ >>> repo = client.defaultRepository
+ >>> repo.id
+ u'83beb297-a6fa-4ac5-844b-98c871c0eea9'
+
+ #. Get the repository's properties. This for-loop spits out everything cmislib knows about the repo.
+
+ >>> repo.name
+ u'Main Repository'
+ >>> info = repo.info
+ >>> for k,v in info.items():
+ ... print "%s:%s" % (k,v)
+ ...
+ cmisSpecificationTitle:Version 1.0 Committee Draft 04
+ cmisVersionSupported:1.0
+ repositoryDescription:None
+ productVersion:3.2.0 (r2 2440)
+ rootFolderId:workspace://SpacesStore/aa1ecedf-9551-49c5-831a-0502bb43f348
+ repositoryId:83beb297-a6fa-4ac5-844b-98c871c0eea9
+ repositoryName:Main Repository
+ vendorName:Alfresco
+ productName:Alfresco Repository (Community)
+
+-------------------
+Folders & Documents
+-------------------
+
+Once you've got the Repository object you can start working with folders.
+
+ #. Create a new folder in the root. You should name yours something unique.
+
+ >>> root = repo.rootFolder
+ >>> someFolder = root.createFolder('someFolder')
+ >>> someFolder.id
+ u'workspace://SpacesStore/91f344ef-84e7-43d8-b379-959c0be7e8fc'
+
+ #. Then, you can create some content:
+
+ >>> someFile = open('test.txt', 'r')
+ >>> someDoc = someFolder.createDocument('Test Document', contentFile=someFile)
+
+ #. And, if you want, you can dump the properties of the newly-created document (this is a partial list):
+
+ >>> props = someDoc.properties
+ >>> for k,v in props.items():
+ ... print '%s:%s' % (k,v)
+ ...
+ cmis:contentStreamMimeType:text/plain
+ cmis:creationDate:2009-12-18T10:59:26.667-06:00
+ cmis:baseTypeId:cmis:document
+ cmis:isLatestMajorVersion:false
+ cmis:isImmutable:false
+ cmis:isMajorVersion:false
+ cmis:objectId:workspace://SpacesStore/2cf36ad5-92b0-4731-94a4-9f3fef25b479
+
+----------------------------------
+Searching For & Retrieving Objects
+----------------------------------
+
+There are several different ways to grab an object:
+ * You can run a CMIS query
+ * You can ask the repository to give you one for a specific path or object ID
+ * You can traverse the repository using a folder's children and/or descendants
+
+ #. Let's find the doc we just created with a full-text search.
+
+ .. note::
+ Note that I'm currently seeing a problem with Alfresco in which the CMIS service returns one less result than what's really there):
+
+ >>> results = repo.query("select * from cmis:document where contains('test')")
+ >>> for result in results:
+ ... print result.name
+ ...
+ Test Document2
+ example test script.js
+
+ #. Alternatively, you can also get objects by their their path, like this:
+
+ >>> someDoc = repo.getObjectByPath('/someFolder/Test Document')
+ >>> someDoc.id
+ u'workspace://SpacesStore/2cf36ad5-92b0-4731-94a4-9f3fef25b479'
+
+ #. Or their object ID, like this:
+
+ >>> someDoc = repo.getObject('workspace://SpacesStore/2cf36ad5-92b0-4731-94a4-9f3fef25b479')
+ >>> someDoc.name
+ u'Test Document'
+
+ #. Folder objects have getChildren() and getDescendants() methods that will return a list of :class:`CmisObject` objects:
+
+ >>> children= someFolder.getChildren()
+ >>> for child in children:
+ ... print child.name
+ ...
+ Test Document
+ Test Document2
Added: incubator/chemistry/trunk/cmislib/src/doc/src/index.rst
URL: http://svn.apache.org/viewvc/incubator/chemistry/trunk/cmislib/src/doc/src/index.rst?rev=912218&view=auto
==============================================================================
--- incubator/chemistry/trunk/cmislib/src/doc/src/index.rst (added)
+++ incubator/chemistry/trunk/cmislib/src/doc/src/index.rst Sat Feb 20 20:32:57 2010
@@ -0,0 +1,29 @@
+.. CMIS Library documentation master file, created by
+ sphinx-quickstart on Thu Dec 10 10:12:43 2009.
+ You can adapt this file completely to your liking, but it should at least
+ contain the root `toctree` directive.
+
+Welcome to CMIS Library's documentation!
+========================================
+
+Contents:
+
+.. toctree::
+ :maxdepth: 2
+
+ about.rst
+ install.rst
+ examples.rst
+ code.rst
+ devguide.rst
+ tests.rst
+ docs.rst
+ sample-data.rst
+
+Indices and tables
+==================
+
+* :ref:`genindex`
+* :ref:`modindex`
+* :ref:`search`
+
Added: incubator/chemistry/trunk/cmislib/src/doc/src/install.rst
URL: http://svn.apache.org/viewvc/incubator/chemistry/trunk/cmislib/src/doc/src/install.rst?rev=912218&view=auto
==============================================================================
--- incubator/chemistry/trunk/cmislib/src/doc/src/install.rst (added)
+++ incubator/chemistry/trunk/cmislib/src/doc/src/install.rst Sat Feb 20 20:32:57 2010
@@ -0,0 +1,23 @@
+Installation
+============
+
+Requirements
+------------
+These requirements must be met:
+ - Python 2.6.x
+ - CMIS provider compliant with CMIS 1.0 Committee Draft 04
+
+ - Alfresco 3.2r2 (`Download <http://wiki.alfresco.com/wiki/Download_Alfresco_Community_Network>`_)
+
+Steps
+-----
+ #. If you don't have `Python <http://www.python.org>`_ installed already, do so.
+ #. If you don't have `setuptools <http://pypi.python.org/pypi/setuptools>`_ installed already, do so.
+ #. Once setuptools is installed, type `easy_install cmislib`
+ #. That's it!
+
+Once you do that, you should be able to fire up Python on the command-line and import cmislib successfully.
+
+ >>> from cmislib.model import CmisClient, Repository, Folder
+
+To validate everything is working, run some :ref:`tests` or walk through some :ref:`examples`.
\ No newline at end of file
Added: incubator/chemistry/trunk/cmislib/src/doc/src/make.bat
URL: http://svn.apache.org/viewvc/incubator/chemistry/trunk/cmislib/src/doc/src/make.bat?rev=912218&view=auto
==============================================================================
--- incubator/chemistry/trunk/cmislib/src/doc/src/make.bat (added)
+++ incubator/chemistry/trunk/cmislib/src/doc/src/make.bat Sat Feb 20 20:32:57 2010
@@ -0,0 +1,113 @@
+@ECHO OFF
+
+REM Command file for Sphinx documentation
+
+set SPHINXBUILD=sphinx-build
+set BUILDDIR=_build
+set ALLSPHINXOPTS=-d %BUILDDIR%/doctrees %SPHINXOPTS% .
+if NOT "%PAPER%" == "" (
+ set ALLSPHINXOPTS=-D latex_paper_size=%PAPER% %ALLSPHINXOPTS%
+)
+
+if "%1" == "" goto help
+
+if "%1" == "help" (
+ :help
+ echo.Please use `make ^<target^>` where ^<target^> is one of
+ echo. html to make standalone HTML files
+ echo. dirhtml to make HTML files named index.html in directories
+ echo. pickle to make pickle files
+ echo. json to make JSON files
+ echo. htmlhelp to make HTML files and a HTML help project
+ echo. qthelp to make HTML files and a qthelp project
+ echo. latex to make LaTeX files, you can set PAPER=a4 or PAPER=letter
+ echo. changes to make an overview over all changed/added/deprecated items
+ echo. linkcheck to check all external links for integrity
+ echo. doctest to run all doctests embedded in the documentation if enabled
+ goto end
+)
+
+if "%1" == "clean" (
+ for /d %%i in (%BUILDDIR%\*) do rmdir /q /s %%i
+ del /q /s %BUILDDIR%\*
+ goto end
+)
+
+if "%1" == "html" (
+ %SPHINXBUILD% -b html %ALLSPHINXOPTS% %BUILDDIR%/html
+ echo.
+ echo.Build finished. The HTML pages are in %BUILDDIR%/html.
+ goto end
+)
+
+if "%1" == "dirhtml" (
+ %SPHINXBUILD% -b dirhtml %ALLSPHINXOPTS% %BUILDDIR%/dirhtml
+ echo.
+ echo.Build finished. The HTML pages are in %BUILDDIR%/dirhtml.
+ goto end
+)
+
+if "%1" == "pickle" (
+ %SPHINXBUILD% -b pickle %ALLSPHINXOPTS% %BUILDDIR%/pickle
+ echo.
+ echo.Build finished; now you can process the pickle files.
+ goto end
+)
+
+if "%1" == "json" (
+ %SPHINXBUILD% -b json %ALLSPHINXOPTS% %BUILDDIR%/json
+ echo.
+ echo.Build finished; now you can process the JSON files.
+ goto end
+)
+
+if "%1" == "htmlhelp" (
+ %SPHINXBUILD% -b htmlhelp %ALLSPHINXOPTS% %BUILDDIR%/htmlhelp
+ echo.
+ echo.Build finished; now you can run HTML Help Workshop with the ^
+.hhp project file in %BUILDDIR%/htmlhelp.
+ goto end
+)
+
+if "%1" == "qthelp" (
+ %SPHINXBUILD% -b qthelp %ALLSPHINXOPTS% %BUILDDIR%/qthelp
+ echo.
+ echo.Build finished; now you can run "qcollectiongenerator" with the ^
+.qhcp project file in %BUILDDIR%/qthelp, like this:
+ echo.^> qcollectiongenerator %BUILDDIR%\qthelp\CMISLibrary.qhcp
+ echo.To view the help file:
+ echo.^> assistant -collectionFile %BUILDDIR%\qthelp\CMISLibrary.ghc
+ goto end
+)
+
+if "%1" == "latex" (
+ %SPHINXBUILD% -b latex %ALLSPHINXOPTS% %BUILDDIR%/latex
+ echo.
+ echo.Build finished; the LaTeX files are in %BUILDDIR%/latex.
+ goto end
+)
+
+if "%1" == "changes" (
+ %SPHINXBUILD% -b changes %ALLSPHINXOPTS% %BUILDDIR%/changes
+ echo.
+ echo.The overview file is in %BUILDDIR%/changes.
+ goto end
+)
+
+if "%1" == "linkcheck" (
+ %SPHINXBUILD% -b linkcheck %ALLSPHINXOPTS% %BUILDDIR%/linkcheck
+ echo.
+ echo.Link check complete; look for any errors in the above output ^
+or in %BUILDDIR%/linkcheck/output.txt.
+ goto end
+)
+
+if "%1" == "doctest" (
+ %SPHINXBUILD% -b doctest %ALLSPHINXOPTS% %BUILDDIR%/doctest
+ echo.
+ echo.Testing of doctests in the sources finished, look at the ^
+results in %BUILDDIR%/doctest/output.txt.
+ goto end
+)
+
+:end
Added: incubator/chemistry/trunk/cmislib/src/doc/src/sample-data.rst
URL: http://svn.apache.org/viewvc/incubator/chemistry/trunk/cmislib/src/doc/src/sample-data.rst?rev=912218&view=auto
==============================================================================
--- incubator/chemistry/trunk/cmislib/src/doc/src/sample-data.rst (added)
+++ incubator/chemistry/trunk/cmislib/src/doc/src/sample-data.rst Sat Feb 20 20:32:57 2010
@@ -0,0 +1,6 @@
+Sample Data
+===========
+
+The /path/to/cmislib/src/data directory contains some sample XML responses from a CMIS service. These are for sample and development purposes and can safely be ignored if you are an end-user of the library.
+
+In some cases there are two files for the same response. For example, 'types.xml' came from Alfresco while 'types.chemistry.xml' came from the simple Apache Chemistry test server.
\ No newline at end of file
Added: incubator/chemistry/trunk/cmislib/src/doc/src/tests.rst
URL: http://svn.apache.org/viewvc/incubator/chemistry/trunk/cmislib/src/doc/src/tests.rst?rev=912218&view=auto
==============================================================================
--- incubator/chemistry/trunk/cmislib/src/doc/src/tests.rst (added)
+++ incubator/chemistry/trunk/cmislib/src/doc/src/tests.rst Sat Feb 20 20:32:57 2010
@@ -0,0 +1,25 @@
+.. _tests:
+
+=====
+Tests
+=====
+
+This code includes unit tests. To run the tests::
+
+ cd /path/to/cmislib/tests
+ Edit settings.py
+ Set REPOSITORY_URL, USERNAME, PASSWORD
+ Optionally, set TEST_ROOT_PATH and other settings to meet your needs
+ python cmislibtest.py
+
+.. note::
+ http://cmis.alfresco.com is a freely-available, hosted CMIS service. If you want to use that for testing, the URL is http://cmis.alfresco.com/s/cmis and the username and password are admin/admin. See the wiki for other known CMIS test servers.
+
+If everything goes well, you should see::
+
+ Ran X tests in 3.607s
+
+ OK
+
+.. note::
+ Until the CMIS specification is ratified, and depending on the implementation of the CMIS provider, you may see errors or failures instead of 'OK'. See the `Known Issues <http://code.google.com/p/cmislib/wiki/KnownIssues>`_ page on the cmislib wiki for a list of known test failures by CMIS provider.
Added: incubator/chemistry/trunk/cmislib/src/tests/250px-Cmis_logo.png
URL: http://svn.apache.org/viewvc/incubator/chemistry/trunk/cmislib/src/tests/250px-Cmis_logo.png?rev=912218&view=auto
==============================================================================
Binary file - no diff available.
Propchange: incubator/chemistry/trunk/cmislib/src/tests/250px-Cmis_logo.png
------------------------------------------------------------------------------
svn:mime-type = application/octet-stream
Added: incubator/chemistry/trunk/cmislib/src/tests/__init__.py
URL: http://svn.apache.org/viewvc/incubator/chemistry/trunk/cmislib/src/tests/__init__.py?rev=912218&view=auto
==============================================================================
(empty)
Added: incubator/chemistry/trunk/cmislib/src/tests/cmislibtest.py
URL: http://svn.apache.org/viewvc/incubator/chemistry/trunk/cmislib/src/tests/cmislibtest.py?rev=912218&view=auto
==============================================================================
--- incubator/chemistry/trunk/cmislib/src/tests/cmislibtest.py (added)
+++ incubator/chemistry/trunk/cmislib/src/tests/cmislibtest.py Sat Feb 20 20:32:57 2010
@@ -0,0 +1,961 @@
+#
+# Licensed 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.
+#
+# Authors:
+# Jeff Potts, Optaros
+#
+'''
+Unit tests for cmislib
+'''
+import unittest
+from cmislib.model import CmisClient
+from cmislib.exceptions import \
+ ObjectNotFoundException, \
+ PermissionDeniedException, \
+ CmisException
+import os
+from time import sleep, time
+import settings
+
+
+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)
+ 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. """
+ self._testFolder.deleteTree()
+
+
+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)
+ self.assert_(cmisClient != 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)
+ 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)
+ repo = cmisClient.getDefaultRepository()
+ self.assert_(repo != None)
+ self.assert_(repo.getRepositoryId() != None)
+
+ def testGetRepository(self):
+ '''Get a repository by repository ID'''
+ cmisClient = CmisClient(settings.REPOSITORY_URL, settings.USERNAME, settings.PASSWORD)
+ 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)
+ self.assertRaises(CmisException, cmisClient.getRepositories)
+
+ def testCmisClientBadAuth(self):
+ '''Try to instantiate a CmisClient object with bad creds'''
+ cmisClient = CmisClient(settings.REPOSITORY_URL, settings.USERNAME, 'BADPASS')
+ self.assertRaises(PermissionDeniedException,
+ cmisClient.getRepositories)
+
+ def testGetRepositoryBadId(self):
+ '''Try to get a repository with a bad repo ID'''
+ cmisClient = CmisClient(settings.REPOSITORY_URL, settings.USERNAME, settings.PASSWORD)
+ 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')
+ self._testContent = self._testFolder.createDocument(testFile.name, contentFile=testFile)
+ testFile.close()
+ testFile = open(settings.TEST_BINARY_2, 'rb')
+ self._testContent2 = self._testFolder.createDocument(settings.TEST_BINARY_2.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'''
+ querySimpleSelect = "SELECT * FROM cmis:document where cmis:name like '" + self._testContent.getProperties()['cmis:name'][:7] + "%'"
+ resultSet = self._repo.query(querySimpleSelect)
+ self.assertTrue(isInResultSet(resultSet, self._testContent))
+
+ def testPropertyMatch(self):
+ '''Find content matching cmis:name property'''
+ querySimpleSelect = "SELECT * FROM cmis:document where cmis:name = '" + self._testContent2.getProperties()['cmis: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 != None)
+ self.assert_(rootFolder.getObjectId() != 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 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')
+ doc10 = self._testFolder.createDocument(settings.TEST_BINARY_1, contentFile=f)
+ doc10Id = doc10.getObjectId()
+ pwc = doc10.checkout()
+ doc11 = pwc.checkin(major='false') # checkin a minor version, 1.1
+ pwc = doc11.checkout()
+ doc20 = pwc.checkin() # checkin a major version, 2.0
+ doc20Id = doc20.getObjectId()
+ 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
+ testFolderName = self._testFolder.getName()
+ parentFolderName = 'testGetObjectByPath folder'
+ subFolderName = 'subfolder'
+ docName = 'testdoc'
+
+ # create the folder structure
+ parentFolder = self._testFolder.createFolder(parentFolderName)
+ subFolder = parentFolder.createFolder(subFolderName)
+ searchFolder = self._repo.getObjectByPath(settings.TEST_ROOT_PATH + "/".join([testFolderName, parentFolderName, subFolderName]))
+ self.assertEquals(subFolder.getObjectId(), searchFolder.getObjectId())
+
+ # create a test doc
+ doc = subFolder.createDocument(docName)
+ searchDoc = self._repo.getObjectByPath(settings.TEST_ROOT_PATH + "/".join([testFolderName, parentFolderName, subFolderName, docName]))
+ self.assertEquals(doc.getObjectId(), searchDoc.getObjectId())
+
+ # get the subfolder by path, then ask for its children
+ subFolder = self._repo.getObjectByPath(settings.TEST_ROOT_PATH + "/".join([testFolderName, parentFolderName, subFolderName]))
+ self.assertEquals(len(subFolder.getChildren().getResults()), 1)
+
+ def testGetUnfiledDocs(self):
+ '''Tests the repository's unfiled collection'''
+
+ if self._repo.getCapabilities()['Unfiling'] != True:
+ 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
+ self.assertFalse(isInResultSet(self._repo.getUnfiledDocs(), newDoc))
+
+ # 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())
+
+ #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 != 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 != 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 != 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 != 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 != 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 != 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 != None)
+ self.assert_('cmis:objectId' in props)
+ self.assert_(props['cmis:objectId'] != None)
+ self.assert_('cmis:objectTypeId' in props)
+ self.assert_(props['cmis:objectTypeId'] != None)
+ self.assert_('cmis:name' in props)
+ self.assert_(props['cmis:name'] != None)
+
+ 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'] != 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())
+
+ # 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')
+
+ 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 DocumentTest(CmisTestBase):
+
+ """ Tests for the :class:`Document` class """
+
+ def testCheckout(self):
+ '''Create a document in a test folder, then check it out'''
+ newDoc = self._testFolder.createDocument('testDocument')
+ pwcDoc = newDoc.checkout()
+ try:
+ self.assertTrue(newDoc.isCheckedOut())
+ self.assert_('cmis:objectId' in newDoc.getProperties())
+ self.assert_('cmis:objectId' in pwcDoc.getProperties())
+ checkedOutDocs = self._repo.getCollection('checkedout')
+ self.assertTrue(isInResultSet(checkedOutDocs, pwcDoc))
+ finally:
+ pwcDoc.delete()
+
+ def testCheckin(self):
+ '''Create a document in a test folder, check it out, then in'''
+ testFilename = settings.TEST_BINARY_1
+ contentFile = open(testFilename, 'rb')
+ testDoc = self._testFolder.createDocument(testFilename, contentFile=contentFile)
+ contentFile.close()
+ self.assertEquals(testFilename, testDoc.getName())
+ 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
+ contentFile = open(testFilename, 'rb')
+ testDoc = self._testFolder.createDocument(testFilename, contentFile=contentFile)
+ contentFile.close()
+ self.assertEquals(testFilename, testDoc.getName())
+ 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'''
+ testFilename = settings.TEST_BINARY_1
+ contentFile = open(testFilename, 'rb')
+ testDoc = self._testFolder.createDocument(testFilename, contentFile=contentFile)
+ 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
+ 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'''
+ newDoc = self._testFolder.createDocument('testDocument')
+ pwcDoc = newDoc.checkout()
+ try:
+ self.assertTrue(newDoc.isCheckedOut())
+ self.assert_('cmis:objectId' in newDoc.getProperties())
+ self.assert_('cmis:objectId' in pwcDoc.getProperties())
+ checkedOutDocs = self._repo.getCollection('checkedout')
+ self.assertTrue(isInResultSet(checkedOutDocs, pwcDoc))
+ finally:
+ pwcDoc.delete()
+ self.assertFalse(newDoc.isCheckedOut())
+ checkedOutDocs = self._repo.getCollection('checkedout')
+ self.assertFalse(isInResultSet(checkedOutDocs, pwcDoc))
+
+ 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')
+ doc10 = self._testFolder.createDocument(settings.TEST_BINARY_1, contentFile=f)
+ pwc = doc10.checkout()
+ doc11 = pwc.checkin(major='false') # checkin a minor version, 1.1
+ pwc = doc11.checkout()
+ doc20 = pwc.checkin() # checkin a major version, 2.0
+ doc20Id = doc20.getObjectId()
+ 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')
+ doc10 = self._testFolder.createDocument(settings.TEST_BINARY_1, contentFile=f)
+ pwc = doc10.checkout()
+ doc11 = pwc.checkin(major='false') # checkin a minor version, 1.1
+ pwc = doc11.checkout()
+ doc20 = pwc.checkin() # checkin a major version, 2.0
+ doc20Label = doc20.getProperties()['cmis:versionLabel']
+ pwc = doc20.checkout()
+ doc21 = pwc.checkin(major='false') # checkin a minor version, 2.1
+ 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', 'cmis:versionLabel': 'foo'}
+ 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
+ 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 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')
+ 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))
+
+ # 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
+
+ # create a test document
+ contentFile = open(settings.TEST_BINARY_1, 'rb')
+ newDoc = self._testFolder.createDocument(settings.TEST_BINARY_1, contentFile=contentFile)
+ contentFile.close()
+
+ 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 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'''
+ testDoc = self._testFolder.createDocument('testdoc')
+ pwc = testDoc.checkout()
+ doc = pwc.checkin() # 2.0
+ pwc = doc.checkout()
+ doc = pwc.checkin() # 3.0
+ 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'])
+
+
+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)
+ repo = cmisClient.getDefaultRepository()
+ typeDefs = repo.getTypeDescendants()
+ found = False
+ for typeDef in typeDefs:
+ if typeDef.getTypeId() == 'cmis:folder':
+ found = True
+ break
+ self.assertTrue(found)
+
+ 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)
+ repo = cmisClient.getDefaultRepository()
+ typeDefs = repo.getTypeChildren()
+ found = False
+ for typeDef in typeDefs:
+ if typeDef.getTypeId() == 'cmis:folder':
+ found = True
+ break
+ self.assertTrue(found)
+
+ 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)
+ 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)
+ 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)
+
+
+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.getResults().has_key(targetDoc.getObjectId()):
+ return True
+ if resultSet.hasNext():
+ resultSet.getNext()
+ else:
+ done = True
+
+if __name__ == "__main__":
+ unittest.main()
Added: incubator/chemistry/trunk/cmislib/src/tests/sample-a.pdf
URL: http://svn.apache.org/viewvc/incubator/chemistry/trunk/cmislib/src/tests/sample-a.pdf?rev=912218&view=auto
==============================================================================
Binary file - no diff available.
Propchange: incubator/chemistry/trunk/cmislib/src/tests/sample-a.pdf
------------------------------------------------------------------------------
svn:executable = *
Propchange: incubator/chemistry/trunk/cmislib/src/tests/sample-a.pdf
------------------------------------------------------------------------------
svn:mime-type = application/octet-stream
Added: incubator/chemistry/trunk/cmislib/src/tests/settings.py
URL: http://svn.apache.org/viewvc/incubator/chemistry/trunk/cmislib/src/tests/settings.py?rev=912218&view=auto
==============================================================================
--- incubator/chemistry/trunk/cmislib/src/tests/settings.py (added)
+++ incubator/chemistry/trunk/cmislib/src/tests/settings.py Sat Feb 20 20:32:57 2010
@@ -0,0 +1,20 @@
+#
+# Override these settings with values to match your environment.
+#
+# CMIS repository's service URL
+REPOSITORY_URL = 'http://cmis.alfresco.com/s/cmis'
+#REPOSITORY_URL = 'http://localhost:8080/alfresco/s/cmis'
+# CMIS repository credentials
+USERNAME = 'admin'
+PASSWORD = 'admin'
+# Absolute path to a directory where test folders can be created, including
+# the trailing slash.
+TEST_ROOT_PATH = '/jeff test/' # REMEMBER TRAILING SLASH
+# Binary test files. Assumed to exist in the same dir as this python script
+TEST_BINARY_1 = '250px-Cmis_logo.png'
+TEST_BINARY_2 = 'sample-a.pdf'
+# For repositories that may index test content asynchronously, the number of
+# times a query is retried before giving up.
+MAX_FULL_TEXT_TRIES = 10
+# The number of seconds the test should sleep between tries.
+FULL_TEXT_WAIT = 10