diff options
| -rw-r--r-- | fatcat/sql.py | 37 | ||||
| -rw-r--r-- | tests/entity_lifecycle.py | 80 |
2 files changed, 117 insertions, 0 deletions
diff --git a/fatcat/sql.py b/fatcat/sql.py index 43b2c2aa..e12757a8 100644 --- a/fatcat/sql.py +++ b/fatcat/sql.py @@ -3,7 +3,9 @@ import json import time import random import hashlib +from sqlalchemy.orm.session import make_transient from fatcat import db +import fatcat.api from fatcat.models import * def populate_db(): @@ -113,3 +115,38 @@ def accept_editgroup(eg): db.session.add(eg.editor) db.session.commit() + +def merge_works(left_id, right_id, edit_group=None): + """Helper to merge two works together.""" + left = WorkIdent.query.filter(WorkIdent.id == left_id).first_or_404() + right = WorkIdent.query.filter(WorkIdent.id == right_id).first_or_404() + assert left.is_live and right.is_live + assert left.rev and right.rev + assert (left.redirect_id == None) and (right.redirect_id == None) + + if edit_group is None: + edit_group = fatcat.api.get_or_create_edit_group() + + releases = ReleaseIdent.query\ + .join(ReleaseIdent.rev)\ + .filter(ReleaseRev.work_ident_id == right_id)\ + .filter(ReleaseIdent.is_live == True)\ + .all() + + # update all right releases to point to left + for release_ident in releases: + rev = release_ident.rev + old_id = rev.id + db.session.expunge(rev) + make_transient(rev) + rev.id = None + rev.parent = old_id + rev.work_ident_id = left.id + re = ReleaseEdit(edit_group=edit_group, ident=release_ident, rev=rev) + db.session.add_all([rev, re]) + + # redirect right id to left (via edit_group) + neww = WorkEdit(edit_group=edit_group, ident=right, + rev=left.rev, redirect_id=left.id) + + db.session.add_all([neww]) diff --git a/tests/entity_lifecycle.py b/tests/entity_lifecycle.py new file mode 100644 index 00000000..9f5b4040 --- /dev/null +++ b/tests/entity_lifecycle.py @@ -0,0 +1,80 @@ + +import json +import pytest +import unittest +import tempfile +import fatcat +import fatcat.sql +from fatcat.models import * +from fixtures import * + + +def test_merge_works(app): + + # two works, each with releases + rv = app.post('/v0/work', + data=json.dumps(dict()), + headers={"content-type": "application/json"}) + workA_id = json.loads(rv.data.decode('utf-8'))['id'] + + rv = app.post('/v0/work', + data=json.dumps(dict()), + headers={"content-type": "application/json"}) + workB_id = json.loads(rv.data.decode('utf-8'))['id'] + + rv = app.post('/v0/release', + data=json.dumps(dict( + title="some release", + work_type="journal-article", + work=workA_id, + doi="10.1234/A1")), + headers={"content-type": "application/json"}) + releaseA1 = json.loads(rv.data.decode('utf-8'))['id'] + + rv = app.post('/v0/release', + data=json.dumps(dict( + title="some release", + work_type="journal-article", + work=workB_id, + doi="10.1234/B1")), + headers={"content-type": "application/json"}) + releaseB1 = json.loads(rv.data.decode('utf-8'))['id'] + + rv = app.post('/v0/release', + data=json.dumps(dict( + title="some release", + work_type="journal-article", + work=workB_id, + doi="10.1234/A1")), + headers={"content-type": "application/json"}) + releaseB2 = json.loads(rv.data.decode('utf-8'))['id'] + + # XXX: what if workB primary was set? + + editgroup_id = 1 + rv = app.post('/v0/editgroup/{}/accept'.format(editgroup_id), + headers={"content-type": "application/json"}) + assert rv.status_code == 200 + assert ChangelogEntry.query.count() == 1 + assert WorkIdent.query.filter(WorkIdent.is_live==True).count() == 2 + assert ReleaseIdent.query.filter(ReleaseIdent.is_live==True).count() == 3 + + # merge works + fatcat.sql.merge_works(workA_id, workB_id) + editgroup_id = 2 + rv = app.post('/v0/editgroup/{}/accept'.format(editgroup_id), + headers={"content-type": "application/json"}) + assert rv.status_code == 200 + + # check results + assert ChangelogEntry.query.count() == 2 + assert WorkIdent.query.filter(WorkIdent.is_live==True).count() == 2 + assert ReleaseIdent.query.filter(ReleaseIdent.is_live==True).count() == 3 + + workA_json = json.loads(app.get('/v0/work/{}'.format(workA_id)).data.decode('utf-8')) + workB_json = json.loads(app.get('/v0/work/{}'.format(workB_id)).data.decode('utf-8')) + assert workA_json['rev'] == workB_json['rev'] + print(workA_json) + print(workB_json) + assert workA_json['redirect_id'] == None + assert workB_json['redirect_id'] == workA_json['id'] |