Commit 61979a02 authored by LE GAC Renaud's avatar LE GAC Renaud
Browse files

Migrate Automaton.

parent eb80c771
This diff is collapsed.
......@@ -469,7 +469,7 @@ class CheckAndFix(object):
value = record.oai()
match = REG_OAI.match(value)
if match.group(2) != record.id():
if int(match.group(2)) != record.id():
db = self.db
# a record with the bad OAI exists in the database
......
......@@ -129,7 +129,7 @@ class Msg(Storage):
self.synonym = record.paper_editor()
if year is None and record is not None:
year = record.year()
year = record.submitted()
self._set(txt, year, translate)
......
# -*- coding: utf-8 -*-
"""test_08_Automaton
* Test Automaton methods
"""
import pytest
from gluon import current
from harvest_tools.automaton import Automaton, ToolException
from harvest_tools.msgcollection import MsgCollection
from invenio_tools.inveniostore import InvenioStore
@pytest.fixture(scope="module")
def svc():
# article for lhcb and forthe current year
return Automaton(current.db,
7,
8,
"article",
2,
year_start=current.request.now.year,
dry_run=True,
debug=True)
def test_process_recid(svc):
"""Test the deepest method to retrieve a record.
"""
# mimic high level stage process_collection and process_url
collection = u"LHCb Papers"
svc.harvester.host = u"cds.cern.ch"
svc.harvester.collections = collection
svc.store = InvenioStore()
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
svc.dbg = True
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
def test_process_collection(svc):
# mimic the previous stage process_url
collection = u"LHCb Papers"
svc.harvester.host = u"cds.cern.ch"
svc.harvester.collections = collection
svc.store = InvenioStore()
# do it
svc.dbg = True
assert svc.process_collection(u"LHCb Papers") is None
# reset
svc.collection_logs = []
svc.logs = []
del svc.harvester.host
del svc.harvester.collections
del svc.store
def test_process_url_cds(svc):
svc.dbg = True
assert svc.process_url("cds.cern.ch", u"LHCb Papers") is None
def test_process_url_ins(svc):
svc.dbg = True
assert svc.process_url(
"inspirehep.net", "find cn lhcb and tc p and not tc c") is None
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment