<s> from __future__ import unicode_literals <EOL> import datetime <EOL> import mock <EOL> import pytest <EOL> from h import db <EOL> from h . api . models import document <EOL> class TestDocumentTitle ( object ) : <EOL> def test_it_returns_the_value_of_the_one_title_DocumentMeta ( self ) : <EOL> """<STR_LIT>""" <EOL> doc = document . Document ( ) <EOL> document . DocumentMeta ( type = '<STR_LIT:title>' , <EOL> value = [ '<STR_LIT>' ] , <EOL> document = doc , <EOL> claimant = '<STR_LIT>' ) <EOL> db . Session . add ( doc ) <EOL> db . Session . flush ( ) <EOL> assert doc . title == '<STR_LIT>' <EOL> def test_it_returns_the_value_of_the_first_title_DocumentMeta ( self ) : <EOL> doc = document . Document ( ) <EOL> document . DocumentMeta ( type = '<STR_LIT:title>' , <EOL> value = [ '<STR_LIT>' ] , <EOL> document = doc , <EOL> claimant = '<STR_LIT>' ) <EOL> document . DocumentMeta ( type = '<STR_LIT:title>' , <EOL> value = [ '<STR_LIT>' ] , <EOL> document = doc , <EOL> claimant = '<STR_LIT>' ) <EOL> db . Session . add ( doc ) <EOL> db . Session . flush ( ) <EOL> assert doc . title == '<STR_LIT>' <EOL> def test_it_returns_None_if_there_are_no_title_DocumentMetas ( self ) : <EOL> doc = document . Document ( ) <EOL> document . DocumentMeta ( type = '<STR_LIT>' , <EOL> value = '<STR_LIT>' , <EOL> document = doc , <EOL> claimant = '<STR_LIT>' ) <EOL> db . Session . add ( doc ) <EOL> db . Session . flush ( ) <EOL> assert doc . title is None <EOL> class TestDocumentFindByURIs ( object ) : <EOL> def test_with_one_matching_Document ( self ) : <EOL> document1 = document . Document ( ) <EOL> uri1 = '<STR_LIT>' <EOL> document1 . document_uris . append ( <EOL> document . DocumentURI ( claimant = uri1 , uri = uri1 ) ) <EOL> document2 = document . Document ( ) <EOL> uri2 = '<STR_LIT>' <EOL> document2 . document_uris . append ( <EOL> document . DocumentURI ( claimant = uri2 , uri = uri2 ) ) <EOL> uri3 = '<STR_LIT>' <EOL> document2 . document_uris . append ( <EOL> document . DocumentURI ( claimant = uri3 , uri = uri2 ) ) <EOL> db . Session . add_all ( [ document1 , document2 ] ) <EOL> db . Session . flush ( ) <EOL> actual = document . Document . find_by_uris ( db . Session , [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] ) <EOL> assert actual . count ( ) == <NUM_LIT:1> <EOL> assert actual . first ( ) == document2 <EOL> def test_no_matches ( self ) : <EOL> document_ = document . Document ( ) <EOL> document_ . document_uris . append ( document . DocumentURI ( <EOL> claimant = '<STR_LIT>' , <EOL> uri = '<STR_LIT>' ) ) <EOL> db . Session . add ( document_ ) <EOL> db . Session . flush ( ) <EOL> actual = document . Document . find_by_uris ( <EOL> db . Session , [ '<STR_LIT>' ] ) <EOL> assert actual . count ( ) == <NUM_LIT:0> <EOL> class TestDocumentFindOrCreateByURIs ( object ) : <EOL> def test_with_one_existing_Document ( self ) : <EOL> """<STR_LIT>""" <EOL> document_ = document . Document ( ) <EOL> docuri1 = document . DocumentURI ( <EOL> claimant = '<STR_LIT>' , <EOL> uri = '<STR_LIT>' , <EOL> document = document_ ) <EOL> docuri2 = document . DocumentURI ( <EOL> claimant = '<STR_LIT>' , <EOL> uri = '<STR_LIT>' , <EOL> document = document_ ) <EOL> db . Session . add ( docuri1 ) <EOL> db . Session . add ( docuri2 ) <EOL> db . Session . flush ( ) <EOL> actual = document . Document . find_or_create_by_uris ( db . Session , <EOL> '<STR_LIT>' , <EOL> [ '<STR_LIT>' , <EOL> '<STR_LIT>' ] ) <EOL> assert actual . count ( ) == <NUM_LIT:1> <EOL> assert actual . first ( ) == document_ <EOL> def test_with_no_existing_documents ( self ) : <EOL> """<STR_LIT>""" <EOL> document_ = document . Document ( ) <EOL> docuri = document . DocumentURI ( <EOL> claimant = '<STR_LIT>' , <EOL> uri = '<STR_LIT>' , <EOL> document = document_ ) <EOL> db . Session . add ( docuri ) <EOL> db . Session . flush ( ) <EOL> documents = document . Document . find_or_create_by_uris ( <EOL> db . Session , <EOL> '<STR_LIT>' , <EOL> [ '<STR_LIT>' ] ) <EOL> assert documents . count ( ) == <NUM_LIT:1> <EOL> actual = documents . first ( ) <EOL> assert isinstance ( actual , document . Document ) <EOL> assert len ( actual . document_uris ) == <NUM_LIT:1> <EOL> docuri = actual . document_uris [ <NUM_LIT:0> ] <EOL> assert docuri . claimant == '<STR_LIT>' <EOL> assert docuri . uri == '<STR_LIT>' <EOL> assert docuri . type == '<STR_LIT>' <EOL> @ pytest . mark . usefixtures ( <EOL> '<STR_LIT>' , <EOL> ) <EOL> class TestCreateOrUpdateDocumentURI ( object ) : <EOL> def test_it_updates_the_existing_DocumentURI_if_there_is_one ( self ) : <EOL> claimant = '<STR_LIT>' <EOL> uri = '<STR_LIT>' <EOL> type_ = '<STR_LIT>' <EOL> content_type = None <EOL> document_ = document . Document ( ) <EOL> created = yesterday ( ) <EOL> updated = yesterday ( ) <EOL> document_uri = document . DocumentURI ( <EOL> claimant = claimant , <EOL> uri = uri , <EOL> type = type_ , <EOL> content_type = content_type , <EOL> document = document_ , <EOL> created = created , <EOL> updated = updated , <EOL> ) <EOL> db . Session . add ( document_uri ) <EOL> now_ = now ( ) <EOL> document . create_or_update_document_uri ( <EOL> session = db . Session , <EOL> claimant = claimant , <EOL> uri = uri , <EOL> type = type_ , <EOL> content_type = content_type , <EOL> document = document_ , <EOL> created = now_ , <EOL> updated = now_ , <EOL> ) <EOL> assert document_uri . created == created <EOL> assert document_uri . updated == now_ <EOL> assert len ( db . Session . query ( document . DocumentURI ) . all ( ) ) == <NUM_LIT:1> , ( <EOL> "<STR_LIT>" ) <EOL> def test_it_creates_a_new_DocumentURI_if_there_is_no_existing_one ( self ) : <EOL> claimant = '<STR_LIT>' <EOL> uri = '<STR_LIT>' <EOL> type_ = '<STR_LIT>' <EOL> content_type = None <EOL> document_ = document . Document ( ) <EOL> created = yesterday ( ) <EOL> updated = yesterday ( ) <EOL> db . Session . add ( document . DocumentURI ( <EOL> claimant = claimant , <EOL> uri = uri , <EOL> type = type_ , <EOL> content_type = '<STR_LIT>' , <EOL> document = document_ , <EOL> created = created , <EOL> updated = updated , <EOL> ) ) <EOL> document . create_or_update_document_uri ( <EOL> session = db . Session , <EOL> claimant = claimant , <EOL> uri = uri , <EOL> type = type_ , <EOL> content_type = content_type , <EOL> document = document_ , <EOL> created = now ( ) , <EOL> updated = now ( ) , <EOL> ) <EOL> document_uri = db . Session . query ( document . DocumentURI ) . all ( ) [ - <NUM_LIT:1> ] <EOL> assert document_uri . claimant == claimant <EOL> assert document_uri . uri == uri <EOL> assert document_uri . type == type_ <EOL> assert document_uri . content_type == content_type <EOL> assert document_uri . document == document_ <EOL> assert document_uri . created > created <EOL> assert document_uri . updated > updated <EOL> def test_it_logs_a_warning_if_document_ids_differ ( self , log , DocumentURI ) : <EOL> """<STR_LIT>""" <EOL> existing_document_uri = mock . Mock ( document = mock_document ( ) ) <EOL> DocumentURI . query . filter . return_value . first . return_value = ( <EOL> existing_document_uri ) <EOL> document . create_or_update_document_uri ( <EOL> session = mock_db_session ( ) , <EOL> claimant = '<STR_LIT>' , <EOL> uri = '<STR_LIT>' , <EOL> type = '<STR_LIT>' , <EOL> content_type = None , <EOL> document = mock_document ( ) , <EOL> created = now ( ) , <EOL> updated = now ( ) ) <EOL> assert log . warn . call_count == <NUM_LIT:1> <EOL> @ pytest . fixture <EOL> def DocumentURI ( self , patch ) : <EOL> return patch ( '<STR_LIT>' ) <EOL> class TestCreateOrUpdateDocumentMeta ( object ) : <EOL> def test_it_creates_a_new_DocumentMeta_if_there_is_no_existing_one ( self ) : <EOL> claimant = '<STR_LIT>' <EOL> type_ = '<STR_LIT:title>' <EOL> value = '<STR_LIT>' <EOL> document_ = document . Document ( ) <EOL> created = yesterday ( ) <EOL> updated = now ( ) <EOL> db . Session . add ( document . DocumentMeta ( <EOL> claimant = claimant , <EOL> type = '<STR_LIT>' , <EOL> value = value , <EOL> document = document_ , <EOL> created = created , <EOL> updated = updated , <EOL> ) ) <EOL> document . create_or_update_document_meta ( <EOL> session = db . Session , <EOL> claimant = claimant , <EOL> type = type_ , <EOL> value = value , <EOL> document = document_ , <EOL> created = created , <EOL> updated = updated , <EOL> ) <EOL> document_meta = db . Session . query ( document . DocumentMeta ) . all ( ) [ - <NUM_LIT:1> ] <EOL> assert document_meta . claimant == claimant <EOL> assert document_meta . type == type_ <EOL> assert document_meta . value == value <EOL> assert document_meta . document == document_ <EOL> assert document_meta . created == created <EOL> assert document_meta . updated == updated <EOL> def test_it_updates_an_existing_DocumentMeta_if_there_is_one ( self ) : <EOL> claimant = '<STR_LIT>' <EOL> type_ = '<STR_LIT:title>' <EOL> value = '<STR_LIT>' <EOL> document_ = document . Document ( ) <EOL> created = yesterday ( ) <EOL> updated = now ( ) <EOL> document_meta = document . DocumentMeta ( <EOL> claimant = claimant , <EOL> type = type_ , <EOL> value = value , <EOL> document = document_ , <EOL> created = created , <EOL> updated = updated , <EOL> ) <EOL> db . Session . add ( document_meta ) <EOL> new_updated = now ( ) <EOL> document . create_or_update_document_meta ( <EOL> session = db . Session , <EOL> claimant = claimant , <EOL> type = type_ , <EOL> value = '<STR_LIT>' , <EOL> document = document . Document ( ) , <EOL> created = now ( ) , <EOL> updated = new_updated , <EOL> ) <EOL> assert document_meta . value == '<STR_LIT>' <EOL> assert document_meta . updated == new_updated <EOL> assert document_meta . created == created , "<STR_LIT>" <EOL> assert document_meta . document == document_ , ( <EOL> "<STR_LIT>" ) <EOL> assert len ( db . Session . query ( document . DocumentMeta ) . all ( ) ) == <NUM_LIT:1> , ( <EOL> "<STR_LIT>" ) <EOL> def test_it_logs_a_warning ( self , DocumentMeta , log ) : <EOL> """<STR_LIT>""" <EOL> document_one = mock_document ( ) <EOL> document_two = mock_document ( ) <EOL> existing_document_meta = mock_document_meta ( document = document_one ) <EOL> DocumentMeta . query . filter . return_value . one_or_none . return_value = existing_document_meta <EOL> document . create_or_update_document_meta ( <EOL> session = mock_db_session ( ) , <EOL> claimant = '<STR_LIT>' , <EOL> type = '<STR_LIT:title>' , <EOL> value = '<STR_LIT>' , <EOL> document = document_two , <EOL> created = yesterday ( ) , <EOL> updated = now ( ) , <EOL> ) <EOL> assert log . warn . call_count == <NUM_LIT:1> <EOL> @ pytest . fixture <EOL> def DocumentMeta ( self , patch ) : <EOL> return patch ( '<STR_LIT>' ) <EOL> @ pytest . mark . usefixtures ( '<STR_LIT>' ) <EOL> class TestMergeDocuments ( object ) : <EOL> def test_merge_documents_returns_master ( self , merge_data ) : <EOL> master , _ = merge_data <EOL> merged_master = document . merge_documents ( db . Session , merge_data ) <EOL> assert merged_master == master <EOL> def test_merge_documents_deletes_duplicate_documents ( self , merge_data ) : <EOL> _ , duplicate = merge_data <EOL> document . merge_documents ( db . Session , merge_data ) <EOL> db . Session . flush ( ) <EOL> assert document . Document . query . get ( duplicate . id ) is None <EOL> def test_merge_documents_rewires_document_uris ( self , merge_data ) : <EOL> master , duplicate = merge_data <EOL> document . merge_documents ( db . Session , merge_data ) <EOL> db . Session . flush ( ) <EOL> assert len ( master . document_uris ) == <NUM_LIT:2> <EOL> assert len ( duplicate . document_uris ) == <NUM_LIT:0> <EOL> def test_merge_documents_rewires_document_meta ( self , merge_data ) : <EOL> master , duplicate = merge_data <EOL> document . merge_documents ( db . Session , merge_data ) <EOL> db . Session . flush ( ) <EOL> assert len ( master . meta ) == <NUM_LIT:2> <EOL> assert len ( duplicate . meta ) == <NUM_LIT:0> <EOL> @ pytest . fixture <EOL> def merge_data ( self , request ) : <EOL> master = document . Document ( document_uris = [ document . DocumentURI ( <EOL> claimant = '<STR_LIT>' , <EOL> uri = '<STR_LIT>' , <EOL> type = '<STR_LIT>' ) ] , <EOL> meta = [ document . DocumentMeta ( <EOL> claimant = '<STR_LIT>' , <EOL> type = '<STR_LIT:title>' , <EOL> value = '<STR_LIT>' ) ] ) <EOL> duplicate = document . Document ( document_uris = [ document . DocumentURI ( <EOL> claimant = '<STR_LIT>' , <EOL> uri = '<STR_LIT>' , <EOL> type = '<STR_LIT>' ) ] , <EOL> meta = [ document . DocumentMeta ( <EOL> claimant = '<STR_LIT>' , <EOL> type = '<STR_LIT:title>' , <EOL> value = '<STR_LIT>' ) ] ) <EOL> db . Session . add_all ( [ master , duplicate ] ) <EOL> db . Session . flush ( ) <EOL> return ( master , duplicate ) <EOL> def now ( ) : <EOL> return datetime . datetime . now ( ) <EOL> def yesterday ( ) : <EOL> return now ( ) - datetime . timedelta ( days = <NUM_LIT:1> ) <EOL> def mock_db_session ( ) : <EOL> """<STR_LIT>""" <EOL> class DB ( object ) : <EOL> def add ( self , obj ) : <EOL> pass <EOL> def query ( self , cls ) : <EOL> pass <EOL> return mock . Mock ( spec = DB ( ) ) <EOL> def mock_document ( ) : <EOL> """<STR_LIT>""" <EOL> return mock . Mock ( spec = document . Document ( ) ) <EOL> def mock_document_meta ( document = None ) : <EOL> class DocumentMeta ( object ) : <EOL> def __init__ ( self ) : <EOL> self . type = None <EOL> self . value = None <EOL> self . created = None <EOL> self . updated = None <EOL> self . document = document <EOL> self . id = None <EOL> self . document_id = None <EOL> return mock . Mock ( spec = DocumentMeta ( ) ) <EOL> @ pytest . fixture <EOL> def log ( patch ) : <EOL> return patch ( '<STR_LIT>' ) </s>
