summaryrefslogtreecommitdiff
path: root/testing/tests/couch/common.py
diff options
context:
space:
mode:
Diffstat (limited to 'testing/tests/couch/common.py')
-rw-r--r--testing/tests/couch/common.py75
1 files changed, 0 insertions, 75 deletions
diff --git a/testing/tests/couch/common.py b/testing/tests/couch/common.py
deleted file mode 100644
index 3c272f6e..00000000
--- a/testing/tests/couch/common.py
+++ /dev/null
@@ -1,75 +0,0 @@
-from uuid import uuid4
-from six.moves.urllib.parse import urljoin
-from couchdb.client import Server
-
-from leap.soledad.common import couch
-from leap.soledad.common.document import ServerDocument
-
-from test_soledad import u1db_tests as tests
-
-
-simple_doc = tests.simple_doc
-nested_doc = tests.nested_doc
-
-
-def make_couch_database_for_test(test, replica_uid):
- dbname = ('test-%s' % uuid4().hex)
- db = couch.CouchDatabase.open_database(
- urljoin(test.couch_url, dbname),
- create=True,
- replica_uid=replica_uid or 'test')
- test.addCleanup(test.delete_db, dbname)
- return db
-
-
-def copy_couch_database_for_test(test, db):
- couch_url = test.couch_url
- new_dbname = db._dbname + '_copy'
- new_db = couch.CouchDatabase.open_database(
- urljoin(couch_url, new_dbname),
- create=True,
- replica_uid=db._replica_uid or 'test')
- # copy all docs
- session = couch.Session()
- old_couch_db = Server(couch_url, session=session)[db._dbname]
- new_couch_db = Server(couch_url, session=session)[new_dbname]
- for doc_id in old_couch_db:
- doc = old_couch_db.get(doc_id)
- # bypass u1db_config document
- if doc_id == 'u1db_config':
- pass
- # copy u1db docs
- elif 'u1db_rev' in doc:
- new_doc = {
- '_id': doc['_id'],
- 'u1db_rev': doc['u1db_rev']
- }
- attachments = []
- if ('u1db_conflicts' in doc):
- new_doc['u1db_conflicts'] = doc['u1db_conflicts']
- for c_rev in doc['u1db_conflicts']:
- attachments.append('u1db_conflict_%s' % c_rev)
- new_couch_db.save(new_doc)
- # save conflict data
- attachments.append('u1db_content')
- for att_name in attachments:
- att = old_couch_db.get_attachment(doc_id, att_name)
- if (att is not None):
- new_couch_db.put_attachment(new_doc, att,
- filename=att_name)
- elif doc_id.startswith('gen-'):
- new_couch_db.save(doc)
- # cleanup connections to prevent file descriptor leaking
- return new_db
-
-
-def make_document_for_test(test, doc_id, rev, content, has_conflicts=False):
- return ServerDocument(
- doc_id, rev, content, has_conflicts=has_conflicts)
-
-
-COUCH_SCENARIOS = [
- ('couch', {'make_database_for_test': make_couch_database_for_test,
- 'copy_database_for_test': copy_couch_database_for_test,
- 'make_document_for_test': make_document_for_test, }),
-]