# -*- coding: utf-8 -*-
###########################################################################
# Copyright (c), The AiiDA team. All rights reserved. #
# This file is part of the AiiDA code. #
# #
# The code is hosted on GitHub at https://github.com/aiidateam/aiida-core #
# For further information on the license, see the LICENSE.txt file #
# For further information please visit http://www.aiida.net #
###########################################################################
"""ORM Log tests"""
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
import logging
from six.moves import range
from aiida import orm
from aiida.backends.testbase import AiidaTestCase
from aiida.common import exceptions
from aiida.common.log import LOG_LEVEL_REPORT
from aiida.common.timezone import now
from aiida.orm import Log
[docs]class TestBackendLog(AiidaTestCase):
"""Test the Log entity"""
[docs] def setUp(self):
super(TestBackendLog, self).setUp()
self.log_record = {
'time': now(),
'loggername': 'loggername',
'levelname': logging.getLevelName(LOG_LEVEL_REPORT),
'dbnode_id': None,
'message': 'This is a template record message',
'metadata': {
'content': 'test'
},
}
[docs] def tearDown(self):
"""
Delete all the created log entries
"""
super(TestBackendLog, self).tearDown()
Log.objects.delete_all()
[docs] def create_log(self):
node = orm.CalculationNode().store()
record = self.log_record
record['dbnode_id'] = node.id
return Log(**record), node
[docs] def test_create_log_message(self):
"""
Test the manual creation of a log entry
"""
entry, node = self.create_log()
self.assertEqual(entry.time, self.log_record['time'])
self.assertEqual(entry.loggername, self.log_record['loggername'])
self.assertEqual(entry.levelname, self.log_record['levelname'])
self.assertEqual(entry.message, self.log_record['message'])
self.assertEqual(entry.metadata, self.log_record['metadata'])
self.assertEqual(entry.dbnode_id, self.log_record['dbnode_id'])
self.assertEqual(entry.dbnode_id, node.id)
[docs] def test_log_delete_single(self):
"""Test that a single log entry can be deleted through the collection."""
entry, _ = self.create_log()
log_id = entry.id
self.assertEqual(len(Log.objects.all()), 1)
# Deleting the entry
Log.objects.delete(log_id)
self.assertEqual(len(Log.objects.all()), 0)
# Deleting a non-existing entry should raise
with self.assertRaises(exceptions.NotExistent):
Log.objects.delete(log_id)
[docs] def test_log_collection_delete_all(self):
"""Test deleting all Log entries through collection"""
count = 10
for _ in range(count):
self.create_log()
log_id = Log.objects.find(limit=1)[0].id
self.assertEqual(len(Log.objects.all()), count)
# Delete all
Log.objects.delete_all()
# Checks
self.assertEqual(len(Log.objects.all()), 0)
with self.assertRaises(exceptions.NotExistent):
Log.objects.delete(log_id)
with self.assertRaises(exceptions.NotExistent):
Log.objects.get(id=log_id)
[docs] def test_log_collection_delete_many(self):
"""Test deleting many Logs through the collection."""
log_ids = []
count = 5
for _ in range(count):
log_, _ = self.create_log()
log_ids.append(log_.id)
special_log, _ = self.create_log()
# Assert the Logs exist
self.assertEqual(len(Log.objects.all()), count + 1)
# Delete new Logs using filter
filters = {'id': {'in': log_ids}}
Log.objects.delete_many(filters=filters)
# Make sure only the special_log Log is left
builder = orm.QueryBuilder().append(Log, project='id')
self.assertEqual(builder.count(), 1)
self.assertEqual(builder.all()[0][0], special_log.id)
for log_id in log_ids:
with self.assertRaises(exceptions.NotExistent):
Log.objects.delete(log_id)
with self.assertRaises(exceptions.NotExistent):
Log.objects.get(id=log_id)
[docs] def test_objects_find(self):
"""Put logs in and find them"""
node = orm.Data().store()
for _ in range(10):
record = self.log_record
record['dbnode_id'] = node.id
Log(**record)
entries = Log.objects.all()
self.assertEqual(10, len(entries))
self.assertIsInstance(entries[0], Log)
[docs] def test_find_orderby(self):
"""
Test the order_by option of log.find
"""
from aiida.orm.logs import OrderSpecifier, ASCENDING, DESCENDING
node_ids = []
for _ in range(10):
_, node = self.create_log()
node_ids.append(node.id)
node_ids.sort()
order_by = [OrderSpecifier('dbnode_id', ASCENDING)]
res_entries = Log.objects.find(order_by=order_by)
self.assertEqual(res_entries[0].dbnode_id, node_ids[0])
order_by = [OrderSpecifier('dbnode_id', DESCENDING)]
res_entries = Log.objects.find(order_by=order_by)
self.assertEqual(res_entries[0].dbnode_id, node_ids[-1])
[docs] def test_find_limit(self):
"""
Test the limit option of log.find
"""
node = orm.Data().store()
limit = 2
for _ in range(limit * 2):
self.log_record['dbnode_id'] = node.id
Log(**self.log_record)
entries = Log.objects.find(limit=limit)
self.assertEqual(len(entries), limit)
[docs] def test_find_filter(self):
"""
Test the filter option of log.find
"""
from random import randint
node_ids = []
for _ in range(10):
_, node = self.create_log()
node_ids.append(node.id)
node_id_of_choice = node_ids.pop(randint(0, 9))
entries = Log.objects.find(filters={'dbnode_id': node_id_of_choice})
self.assertEqual(len(entries), 1)
self.assertEqual(entries[0].dbnode_id, node_id_of_choice)
[docs] def test_db_log_handler(self):
"""
Verify that the db log handler is attached correctly
by firing a log message through the regular logging module
attached to a calculation node
"""
from aiida.orm.logs import OrderSpecifier, ASCENDING
message = 'Testing logging of critical failure'
node = orm.CalculationNode()
# Firing a log for an unstored should not end up in the database
node.logger.critical(message)
logs = Log.objects.find()
self.assertEqual(len(logs), 0)
# After storing the node, logs above log level should be stored
node.store()
node.logger.critical(message)
logs = Log.objects.find()
self.assertEqual(len(logs), 1)
self.assertEqual(logs[0].message, message)
# Launching a second log message ensuring that both messages are correctly stored
message2 = message + ' - Second message'
node.logger.critical(message2)
order_by = [OrderSpecifier('time', ASCENDING)]
logs = Log.objects.find(order_by=order_by)
self.assertEqual(len(logs), 2)
self.assertEqual(logs[0].message, message)
self.assertEqual(logs[1].message, message2)
[docs] def test_log_querybuilder(self):
""" Test querying for logs by joining on nodes in the QueryBuilder """
from aiida.orm import QueryBuilder
# Setup nodes
log_1, calc = self.create_log()
log_2 = Log(now(), 'loggername', logging.getLevelName(LOG_LEVEL_REPORT), calc.id, 'log message #2')
log_3 = Log(now(), 'loggername', logging.getLevelName(LOG_LEVEL_REPORT), calc.id, 'log message #3')
# Retrieve a node by joining on a specific log ('log_1')
builder = QueryBuilder()
builder.append(Log, tag='log', filters={'id': log_2.id})
builder.append(orm.CalculationNode, with_log='log', project=['uuid'])
nodes = builder.all()
self.assertEqual(len(nodes), 1)
for node in nodes:
self.assertIn(str(node[0]), [calc.uuid])
# Retrieve all logs for a specific node by joining on a said node
builder = QueryBuilder()
builder.append(orm.CalculationNode, tag='calc', filters={'id': calc.id})
builder.append(Log, with_node='calc', project=['uuid'])
logs = builder.all()
self.assertEqual(len(logs), 3)
for log in logs:
self.assertIn(str(log[0]), [str(log_1.uuid), str(log_2.uuid), str(log_3.uuid)])