test_09_Automaton.py 3.2 KB
Newer Older
LE GAC Renaud's avatar
LE GAC Renaud committed
1 2 3 4 5 6
"""test_08_Automaton

    * Test Automaton methods

"""
import pytest
LE GAC Renaud's avatar
LE GAC Renaud committed
7

LE GAC Renaud's avatar
LE GAC Renaud committed
8
from gluon import current
LE GAC Renaud's avatar
LE GAC Renaud committed
9
from harvest_tools.automaton import Automaton
LE GAC Renaud's avatar
LE GAC Renaud committed
10
from harvest_tools.msgcollection import MsgCollection
11
from plugin_dbui import get_id
12
from store_tools.factory import build_store
LE GAC Renaud's avatar
LE GAC Renaud committed
13 14 15 16


@pytest.fixture(scope="module")
def svc():
17
    db = current.db
LE GAC Renaud's avatar
LE GAC Renaud committed
18
    # article for lhcb and for the current year
19 20 21
    return Automaton(db,
                     get_id(db.teams, team="LHCb"),
                     get_id(db.projects, project="LHCb"),
LE GAC Renaud's avatar
LE GAC Renaud committed
22
                     "article",
23
                     get_id(db.categories, code="ACL"),
LE GAC Renaud's avatar
LE GAC Renaud committed
24
                     year_start=current.request.now.year,
25
                     dry_run=True)
LE GAC Renaud's avatar
LE GAC Renaud committed
26 27


28
def test__is_record_in_db_09001(svc):
LE GAC Renaud's avatar
LE GAC Renaud committed
29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50
    """The record cds1389970 and ins939619 describe the same LHCb paper.

        * the paper is loaded in the database from cds.cern.ch.
        * primary and secondary OAI are defined

    """
    db = current.db
    print

    # check that the record is in the database
    query = \
        db.publications.origin.contains("http://cds.cern.ch/record/1389907")

    row = db(query).select().first()

    assert row.origin == \
        "http://cds.cern.ch/record/1389907, " \
        "http://inspirehep.net/record/939619"

    rec_id_1 = svc._is_record_in_db(
        "Luminosity measurement",
        host="cds.cern.ch",
51
        rec_id=1389907,)
LE GAC Renaud's avatar
LE GAC Renaud committed
52 53 54 55 56 57 58 59 60 61 62

    assert rec_id_1 != 0

    rec_id_2 = svc._is_record_in_db(
        "Luminosity measurement",
        oai_url="http://inspirehep.net/record/939619")

    assert rec_id_2 != 0
    assert rec_id_1 == rec_id_2


63
def test_process_recid_09002(svc):
LE GAC Renaud's avatar
LE GAC Renaud committed
64 65 66
    """Test the deepest method to retrieve a record.

    """
LE GAC Renaud's avatar
LE GAC Renaud committed
67 68 69 70
    # reset
    svc.collection_logs = []
    svc.logs = []

LE GAC Renaud's avatar
LE GAC Renaud committed
71
    # mimic high level stage process_collection and process_url
LE GAC Renaud's avatar
LE GAC Renaud committed
72
    collection = "LHCb Papers"
LE GAC Renaud's avatar
LE GAC Renaud committed
73

LE GAC Renaud's avatar
LE GAC Renaud committed
74
    svc.harvester.host = "cds.cern.ch"
LE GAC Renaud's avatar
LE GAC Renaud committed
75
    svc.harvester.collections = collection
76
    svc.store = build_store(host="cds.cern.ch")
LE GAC Renaud's avatar
LE GAC Renaud committed
77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101

    ctitle = "LHCb / article / %s" % collection
    svc.collection_logs.append(MsgCollection(title=ctitle))

    # get a list of ids
    kwargs = svc._search_parameters(collection)
    recids = svc.store.get_ids(**kwargs)
    assert len(recids) > 0

    # try with the oldest one
    recid = recids[-1]
    assert svc.process_recid(recid) is None

    assert len(svc.logs) == 1
    assert svc.logs[-1].action is None
    assert svc.logs[-1].txt is None

    # reset
    svc.collection_logs = []
    svc.logs = []
    del svc.harvester.host
    del svc.harvester.collections
    del svc.store


102
def test_process_collection_09003(svc):
LE GAC Renaud's avatar
LE GAC Renaud committed
103 104

    # mimic the previous stage process_url
LE GAC Renaud's avatar
LE GAC Renaud committed
105
    collection = "LHCb Papers"
LE GAC Renaud's avatar
LE GAC Renaud committed
106

LE GAC Renaud's avatar
LE GAC Renaud committed
107
    svc.harvester.host = "cds.cern.ch"
LE GAC Renaud's avatar
LE GAC Renaud committed
108
    svc.harvester.collections = collection
109
    svc.store = build_store(host="cds.cern.ch")
LE GAC Renaud's avatar
LE GAC Renaud committed
110 111

    # do it
LE GAC Renaud's avatar
LE GAC Renaud committed
112
    assert svc.process_collection("LHCb Papers") is None
LE GAC Renaud's avatar
LE GAC Renaud committed
113 114 115 116 117 118 119 120 121

    # reset
    svc.collection_logs = []
    svc.logs = []
    del svc.harvester.host
    del svc.harvester.collections
    del svc.store


122
def test_process_url_cds_09004(svc):
LE GAC Renaud's avatar
LE GAC Renaud committed
123
    assert svc.process_url("cds.cern.ch", "LHCb Papers") is None
LE GAC Renaud's avatar
LE GAC Renaud committed
124 125


126
def test_process_url_ins_09005(svc):
LE GAC Renaud's avatar
LE GAC Renaud committed
127 128
    assert svc.process_url(
        "inspirehep.net", "find cn lhcb and tc p and not tc c") is None