465 lines
14 KiB
Python
465 lines
14 KiB
Python
# This file is part of Tryton. The COPYRIGHT file at the top level of
|
|
# this repository contains the full copyright notices and license terms.
|
|
import unittest
|
|
import datetime
|
|
|
|
from trytond.tests.test_tryton import install_module, with_transaction
|
|
from trytond.transaction import Transaction
|
|
from trytond.pool import Pool
|
|
from trytond.exceptions import UserError
|
|
from trytond import backend
|
|
|
|
|
|
class HistoryTestCase(unittest.TestCase):
|
|
'Test History'
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
install_module('tests')
|
|
|
|
@with_transaction()
|
|
def tearDown(self):
|
|
pool = Pool()
|
|
History = pool.get('test.history')
|
|
transaction = Transaction()
|
|
cursor = transaction.connection.cursor()
|
|
table = History.__table__()
|
|
history_table = History.__table_history__()
|
|
cursor.execute(*table.delete())
|
|
cursor.execute(*history_table.delete())
|
|
transaction.commit()
|
|
|
|
@with_transaction()
|
|
def test_read(self):
|
|
'Test read history'
|
|
pool = Pool()
|
|
History = pool.get('test.history')
|
|
transaction = Transaction()
|
|
|
|
# Create some history entry
|
|
# It is needed to commit to have different timestamps
|
|
history = History(value=1)
|
|
history.save()
|
|
history_id = history.id
|
|
first = history.create_date
|
|
|
|
transaction.commit()
|
|
|
|
history = History(history_id)
|
|
history.value = 2
|
|
history.save()
|
|
second = history.write_date
|
|
|
|
transaction.commit()
|
|
|
|
history = History(history_id)
|
|
history.value = 3
|
|
history.save()
|
|
third = history.write_date
|
|
|
|
transaction.commit()
|
|
|
|
for timestamp, value in [
|
|
(first, 1),
|
|
(second, 2),
|
|
(third, 3),
|
|
(datetime.datetime.now(), 3),
|
|
(datetime.datetime.max, 3),
|
|
]:
|
|
with Transaction().set_context(_datetime=timestamp):
|
|
history = History(history_id)
|
|
self.assertEqual(history.value, value)
|
|
|
|
with Transaction().set_context(_datetime=datetime.datetime.min):
|
|
self.assertRaises(UserError, History.read, [history_id])
|
|
|
|
@unittest.skipUnless(backend.name() == 'postgresql',
|
|
'CURRENT_TIMESTAMP as transaction_timestamp is specific to postgresql')
|
|
@with_transaction()
|
|
def test_read_same_timestamp(self):
|
|
'Test read history with same timestamp'
|
|
pool = Pool()
|
|
History = pool.get('test.history')
|
|
transaction = Transaction()
|
|
|
|
history = History(value=1)
|
|
history.save()
|
|
history_id = history.id
|
|
first = history.create_date
|
|
|
|
history.value = 2
|
|
history.save()
|
|
second = history.write_date
|
|
|
|
self.assertEqual(first, second)
|
|
|
|
transaction.commit()
|
|
|
|
history = History(history_id)
|
|
history.value = 3
|
|
history.save()
|
|
third = history.write_date
|
|
|
|
transaction.commit()
|
|
|
|
for timestamp, value in [
|
|
(first, 2),
|
|
(third, 3),
|
|
]:
|
|
with Transaction().set_context(_datetime=timestamp):
|
|
history = History(history_id)
|
|
self.assertEqual(history.value, value)
|
|
|
|
@with_transaction()
|
|
def test_history_revisions(self):
|
|
'Test history revisions'
|
|
pool = Pool()
|
|
History = pool.get('test.history')
|
|
transaction = Transaction()
|
|
|
|
history = History(value=1)
|
|
history.save()
|
|
history_id = history.id
|
|
first = history.create_date
|
|
|
|
transaction.commit()
|
|
|
|
history = History(history_id)
|
|
history.value = 2
|
|
history.save()
|
|
second = history.write_date
|
|
|
|
transaction.commit()
|
|
|
|
history = History(history_id)
|
|
history.value = 3
|
|
history.save()
|
|
third = history.write_date
|
|
|
|
transaction.commit()
|
|
|
|
revisions = History.history_revisions([history_id])
|
|
self.assertEqual(revisions, [
|
|
(third, history_id, u'Administrator'),
|
|
(second, history_id, u'Administrator'),
|
|
(first, history_id, u'Administrator'),
|
|
])
|
|
|
|
@with_transaction()
|
|
def test_restore_history(self):
|
|
'Test restore history'
|
|
pool = Pool()
|
|
History = pool.get('test.history')
|
|
transaction = Transaction()
|
|
|
|
history = History(value=1)
|
|
history.save()
|
|
history_id = history.id
|
|
first = history.create_date
|
|
|
|
transaction.commit()
|
|
|
|
history = History(history_id)
|
|
history.value = 2
|
|
history.save()
|
|
|
|
transaction.commit()
|
|
|
|
History.restore_history([history_id], first)
|
|
history = History(history_id)
|
|
self.assertEqual(history.value, 1)
|
|
|
|
transaction.rollback()
|
|
|
|
History.restore_history([history_id], datetime.datetime.min)
|
|
self.assertRaises(UserError, History.read, [history_id])
|
|
|
|
transaction.rollback()
|
|
|
|
History.delete([History(history_id)])
|
|
|
|
transaction.commit()
|
|
|
|
History.restore_history([history_id], datetime.datetime.max)
|
|
self.assertRaises(UserError, History.read, [history_id])
|
|
|
|
@with_transaction()
|
|
def test_restore_history_before(self):
|
|
'Test restore history before'
|
|
pool = Pool()
|
|
History = pool.get('test.history')
|
|
transaction = Transaction()
|
|
|
|
history = History(value=1)
|
|
history.save()
|
|
history_id = history.id
|
|
|
|
transaction.commit()
|
|
|
|
history = History(history_id)
|
|
history.value = 2
|
|
history.save()
|
|
second = history.write_date
|
|
|
|
transaction.commit()
|
|
|
|
history = History(history_id)
|
|
history.value = 3
|
|
history.save()
|
|
|
|
transaction.commit()
|
|
|
|
History.restore_history_before([history_id], second)
|
|
history = History(history_id)
|
|
self.assertEqual(history.value, 1)
|
|
|
|
@unittest.skipUnless(backend.name() == 'postgresql',
|
|
'CURRENT_TIMESTAMP as transaction_timestamp is specific to postgresql')
|
|
@with_transaction()
|
|
def test_restore_history_same_timestamp(self):
|
|
'Test restore history with same timestamp'
|
|
pool = Pool()
|
|
History = pool.get('test.history')
|
|
transaction = Transaction()
|
|
|
|
history = History(value=1)
|
|
history.save()
|
|
history_id = history.id
|
|
first = history.create_date
|
|
history.value = 2
|
|
history.save()
|
|
second = history.create_date
|
|
|
|
self.assertEqual(first, second)
|
|
|
|
transaction.commit()
|
|
|
|
history = History(history_id)
|
|
history.value = 3
|
|
history.save()
|
|
|
|
transaction.commit()
|
|
|
|
History.restore_history([history_id], first)
|
|
history = History(history_id)
|
|
self.assertEqual(history.value, 2)
|
|
|
|
@with_transaction()
|
|
def test_ordered_search(self):
|
|
'Test ordered search of history models'
|
|
pool = Pool()
|
|
History = pool.get('test.history')
|
|
transaction = Transaction()
|
|
order = [('value', 'ASC')]
|
|
|
|
history = History(value=1)
|
|
history.save()
|
|
first_id = history.id
|
|
first_stamp = history.create_date
|
|
transaction.commit()
|
|
|
|
history = History(value=2)
|
|
history.save()
|
|
second_id = history.id
|
|
second_stamp = history.create_date
|
|
|
|
transaction.commit()
|
|
|
|
first, second = History.search([], order=order)
|
|
|
|
self.assertEqual(first.id, first_id)
|
|
self.assertEqual(second.id, second_id)
|
|
|
|
first.value = 3
|
|
first.save()
|
|
third_stamp = first.write_date
|
|
transaction.commit()
|
|
|
|
results = [
|
|
(first_stamp, [first]),
|
|
(second_stamp, [first, second]),
|
|
(third_stamp, [second, first]),
|
|
(datetime.datetime.now(), [second, first]),
|
|
(datetime.datetime.max, [second, first]),
|
|
]
|
|
for timestamp, instances in results:
|
|
with Transaction().set_context(_datetime=timestamp):
|
|
records = History.search([], order=order)
|
|
self.assertEqual(records, instances)
|
|
transaction.rollback()
|
|
|
|
to_delete, _ = History.search([], order=order)
|
|
|
|
self.assertEqual(to_delete.id, second.id)
|
|
|
|
History.delete([to_delete])
|
|
transaction.commit()
|
|
|
|
results = [
|
|
(first_stamp, [first]),
|
|
(second_stamp, [first, second]),
|
|
(third_stamp, [second, first]),
|
|
(datetime.datetime.now(), [first]),
|
|
(datetime.datetime.max, [first]),
|
|
]
|
|
for timestamp, instances in results:
|
|
with Transaction().set_context(_datetime=timestamp,
|
|
from_test=True):
|
|
records = History.search([], order=order)
|
|
self.assertEqual(records, instances)
|
|
transaction.rollback()
|
|
|
|
@unittest.skipUnless(backend.name() == 'postgresql',
|
|
'CURRENT_TIMESTAMP as transaction_timestamp is specific to postgresql')
|
|
@with_transaction()
|
|
def test_ordered_search_same_timestamp(self):
|
|
'Test ordered search with same timestamp'
|
|
pool = Pool()
|
|
History = pool.get('test.history')
|
|
transaction = Transaction()
|
|
order = [('value', 'ASC')]
|
|
|
|
history = History(value=1)
|
|
history.save()
|
|
first_stamp = history.create_date
|
|
history.value = 4
|
|
history.save()
|
|
second_stamp = history.write_date
|
|
|
|
self.assertEqual(first_stamp, second_stamp)
|
|
transaction.commit()
|
|
|
|
results = [
|
|
(second_stamp, [history], [4]),
|
|
(datetime.datetime.now(), [history], [4]),
|
|
(datetime.datetime.max, [history], [4]),
|
|
]
|
|
|
|
for timestamp, instances, values in results:
|
|
with Transaction().set_context(_datetime=timestamp,
|
|
last_test=True):
|
|
records = History.search([], order=order)
|
|
self.assertEqual(records, instances)
|
|
self.assertEqual([x.value for x in records], values)
|
|
transaction.rollback()
|
|
|
|
@with_transaction()
|
|
def test_browse(self):
|
|
'Test browsing history'
|
|
pool = Pool()
|
|
History = pool.get('test.history')
|
|
Line = pool.get('test.history.line')
|
|
transaction = Transaction()
|
|
|
|
history = History(value=1)
|
|
history.save()
|
|
history_id = history.id
|
|
line_a = Line(name='a', history=history)
|
|
line_a.save()
|
|
line_a_id = line_a.id
|
|
line_b = Line(name='b', history=history)
|
|
line_b.save()
|
|
line_b_id = line_b.id
|
|
|
|
first_stamp = line_b.create_date
|
|
|
|
history.stamp = first_stamp
|
|
history.save()
|
|
|
|
transaction.commit()
|
|
|
|
history = History(history_id)
|
|
history.value = 2
|
|
history.save()
|
|
|
|
Line.delete([Line(line_b_id)])
|
|
|
|
line_a = Line(line_a_id)
|
|
line_a.name = 'c'
|
|
line_a.save()
|
|
|
|
second_stamp = line_a.write_date
|
|
|
|
transaction.commit()
|
|
|
|
history = History(history_id)
|
|
self.assertEqual(history.value, 2)
|
|
self.assertEqual([l.name for l in history.lines], ['c'])
|
|
self.assertEqual(history.stamp, first_stamp)
|
|
self.assertEqual(
|
|
[l.name for l in history.lines_at_stamp], ['a', 'b'])
|
|
|
|
with Transaction().set_context(_datetime=first_stamp):
|
|
history = History(history_id)
|
|
self.assertEqual(history.value, 1)
|
|
self.assertEqual([l.name for l in history.lines], ['a', 'b'])
|
|
|
|
with Transaction().set_context(_datetime=second_stamp):
|
|
history = History(history_id)
|
|
self.assertEqual(history.value, 2)
|
|
self.assertEqual([l.name for l in history.lines], ['c'])
|
|
self.assertEqual(history.stamp, first_stamp)
|
|
self.assertEqual(
|
|
[l.name for l in history.lines_at_stamp], ['a', 'b'])
|
|
|
|
@with_transaction()
|
|
def test_search_cursor_max(self):
|
|
'Test search with number of history entries at database.IN_MAX'
|
|
pool = Pool()
|
|
History = pool.get('test.history')
|
|
transaction = Transaction()
|
|
database = transaction.database
|
|
|
|
history = History(value=-1)
|
|
history.save()
|
|
|
|
for history.value in range(database.IN_MAX + 1):
|
|
history.save()
|
|
|
|
with transaction.set_context(_datetime=datetime.datetime.max):
|
|
record, = History.search([])
|
|
|
|
self.assertEqual(record.value, database.IN_MAX)
|
|
|
|
@with_transaction()
|
|
def test_search_cursor_max_entries(self):
|
|
'Test search for skipping first history entries at database.IN_MAX'
|
|
pool = Pool()
|
|
History = pool.get('test.history')
|
|
transaction = Transaction()
|
|
database = transaction.database
|
|
|
|
for i in xrange(0, 2):
|
|
history = History(value=-1)
|
|
history.save()
|
|
|
|
for history.value in range(database.IN_MAX + 1):
|
|
history.save()
|
|
|
|
with transaction.set_context(_datetime=datetime.datetime.max):
|
|
records = History.search([])
|
|
|
|
self.assertEqual({r.value for r in records}, {database.IN_MAX})
|
|
self.assertEqual(len(records), 2)
|
|
|
|
@with_transaction()
|
|
def test_search_cursor_max_histories(self):
|
|
'Test search with number of histories at database.IN_MAX'
|
|
pool = Pool()
|
|
History = pool.get('test.history')
|
|
transaction = Transaction()
|
|
database = transaction.database
|
|
|
|
n = database.IN_MAX + 1
|
|
History.create([{'value': 1}] * n)
|
|
|
|
with transaction.set_context(_datetime=datetime.datetime.max):
|
|
records = History.search([])
|
|
|
|
self.assertEqual({r.value for r in records}, {1})
|
|
self.assertEqual(len(records), n)
|
|
|
|
|
|
def suite():
|
|
return unittest.TestLoader().loadTestsFromTestCase(HistoryTestCase)
|