HCE Project Python language Distributed Tasks Manager Application, Distributed Crawler Application and client API bindings.  2.0.0-chaika
Hierarchical Cluster Engine Python language binding
tests.test_dtm_TaskLogScheme.TestTaskLogScheme Class Reference
Inheritance diagram for tests.test_dtm_TaskLogScheme.TestTaskLogScheme:
Collaboration diagram for tests.test_dtm_TaskLogScheme.TestTaskLogScheme:

Public Member Functions

def setUp (self)
 
def tearDown (self)
 
def getNewTaskLog (self, startValue)
 
def checkError (self)
 
def test_create_empty_db (self)
 
def test_insert_data (self)
 
def test_del_data (self)
 
def test_update_data (self)
 
def test_insert_in_two_bases (self)
 

Public Attributes

 dbi
 

Detailed Description

Definition at line 15 of file test_dtm_TaskLogScheme.py.

Member Function Documentation

◆ checkError()

def tests.test_dtm_TaskLogScheme.TestTaskLogScheme.checkError (   self)

Definition at line 48 of file test_dtm_TaskLogScheme.py.

48  def checkError(self):
49  print self.dbi.getErrorCode()
50  if self.dbi.getErrorCode() != dbi_consts.DBI_SUCCESS_CODE:
51  assert(False)
52 
53 
Here is the caller graph for this function:

◆ getNewTaskLog()

def tests.test_dtm_TaskLogScheme.TestTaskLogScheme.getNewTaskLog (   self,
  startValue 
)

Definition at line 28 of file test_dtm_TaskLogScheme.py.

28  def getNewTaskLog(self, startValue):
29  taskLog = TaskLog()
30  taskLog.id = startValue
31  taskLog.pId = startValue + 1
32  taskLog.nodeName =str(startValue + 2)
33  taskLog.cDate = datetime.datetime.now()
34  taskLog.sDate = datetime.datetime.now()
35  taskLog.rDate = datetime.datetime.now()
36  taskLog.fDate = datetime.datetime.now()
37  taskLog.pTime = startValue + 7
38  taskLog.pTimeMax = startValue + 8
39  taskLog.state = startValue + 9
40  taskLog.uRRAM = startValue + 10
41  taskLog.uVRAM = startValue + 11
42  taskLog.uCPU = startValue + 12
43  taskLog.uThreads = startValue + 13
44  taskLog.tries = startValue + 14
45  return taskLog
46 
47 
Here is the caller graph for this function:

◆ setUp()

def tests.test_dtm_TaskLogScheme.TestTaskLogScheme.setUp (   self)

Definition at line 18 of file test_dtm_TaskLogScheme.py.

18  def setUp(self):
19  config_dbi = dict({"db_name": "sqlite:///:memory:"})
20  self.dbi = DBI(config_dbi)
21 
22 

◆ tearDown()

def tests.test_dtm_TaskLogScheme.TestTaskLogScheme.tearDown (   self)

Definition at line 23 of file test_dtm_TaskLogScheme.py.

23  def tearDown(self):
24  del self.dbi
25  self.dbi = None
26 
27 

◆ test_create_empty_db()

def tests.test_dtm_TaskLogScheme.TestTaskLogScheme.test_create_empty_db (   self)

Definition at line 54 of file test_dtm_TaskLogScheme.py.

54  def test_create_empty_db(self):
55  taskLog = self.getNewTaskLog(1)
56  taskLogScheme = TaskLogScheme(taskLog)
57  results = self.dbi.fetchAll(taskLogScheme)
58  self.checkError()
59 
60 
61  self.assertEqual(len(results[0]), 0, "")
62 
63 
Here is the call graph for this function:

◆ test_del_data()

def tests.test_dtm_TaskLogScheme.TestTaskLogScheme.test_del_data (   self)

Definition at line 84 of file test_dtm_TaskLogScheme.py.

84  def test_del_data(self):
85  taskLog = self.getNewTaskLog(1)
86  taskLogScheme = TaskLogScheme(taskLog)
87 
88  self.dbi.insert(taskLogScheme)
89  self.checkError()
90  self.dbi.delete(taskLogScheme, "id=%s" % taskLog.id)
91  self.checkError()
92  results = self.dbi.fetchAll(taskLogScheme)
93  self.checkError()
94 
95  self.assertEqual(len(results[0]), 0, "")
96 
97 
Here is the call graph for this function:

◆ test_insert_data()

def tests.test_dtm_TaskLogScheme.TestTaskLogScheme.test_insert_data (   self)

Definition at line 64 of file test_dtm_TaskLogScheme.py.

64  def test_insert_data(self):
65  taskLog = self.getNewTaskLog(1)
66  taskLogScheme = TaskLogScheme(taskLog)
67 
68  lookTaskLog = TaskLog()
69  lookTaskLog.id = 1
70  lookTaskLogScheme = TaskLogScheme(lookTaskLog)
71 
72  self.dbi.insert(taskLogScheme)
73  self.checkError()
74 
75  fetchTaskLogScheme = self.dbi.fetch(lookTaskLogScheme, "id=%s" % taskLog.id)[0]
76  self.checkError()
77  fetchTaskLog = fetchTaskLogScheme._getTaskLog()
78 
79  self.assertEqual(taskLog.__dict__, fetchTaskLog.__dict__, "")
80 
81  self.dbi.delete(taskLogScheme, "id=%s" % taskLog.id)
82 
83 
Here is the call graph for this function:

◆ test_insert_in_two_bases()

def tests.test_dtm_TaskLogScheme.TestTaskLogScheme.test_insert_in_two_bases (   self)

Definition at line 118 of file test_dtm_TaskLogScheme.py.

118  def test_insert_in_two_bases(self):
119  taskLog = self.getNewTaskLog(1)
120  taskLogScheme = TaskLogScheme(taskLog)
121  taskBackLogScheme = TaskBackLogScheme(taskLog)
122 
123  self.dbi.insert(taskLogScheme)
124  self.checkError()
125 
126  self.dbi.insert(taskBackLogScheme)
127  self.checkError()
128 
129  fetchTaskLogScheme = self.dbi.fetch(taskLogScheme, "id=%s" % taskLog.id)[0]
130  self.checkError()
131  fetchTaskLog = fetchTaskLogScheme._getTaskLog()
132 
133  self.assertEqual(taskLog.__dict__, fetchTaskLog.__dict__, "")
134 
135  fetchTaskBackLogScheme = self.dbi.fetch(taskBackLogScheme, "id=%s" % taskLog.id)[0]
136  self.checkError()
137  fetchTaskBackLog = fetchTaskBackLogScheme._getTaskLog()
138 
139  self.assertEqual(taskLog.__dict__, fetchTaskBackLog.__dict__, "")
140 
141  #clean up
142  self.dbi.delete(taskLogScheme, "id=%s" % taskLog.id)
143  self.dbi.delete(taskBackLogScheme, "id=%s" % taskLog.id)
144 
145 
Here is the call graph for this function:

◆ test_update_data()

def tests.test_dtm_TaskLogScheme.TestTaskLogScheme.test_update_data (   self)

Definition at line 98 of file test_dtm_TaskLogScheme.py.

98  def test_update_data(self):
99  taskLog = self.getNewTaskLog(1)
100  taskLogScheme = TaskLogScheme(taskLog)
101 
102  self.dbi.insert(taskLogScheme)
103  self.checkError()
104 
105  taskLog.pId = 101
106  taskLogScheme = TaskLogScheme(taskLog)
107  self.dbi.update(taskLogScheme, "id=%s" % taskLog.id)
108  self.checkError()
109 
110  fetchTaskLogScheme = self.dbi.fetch(taskLogScheme, "id=%s" % taskLog.id)[0]
111  fetchTaskLog = fetchTaskLogScheme._getTaskLog()
112 
113  self.assertEqual(fetchTaskLog.pId, taskLog.pId, "")
114 
115  self.dbi.delete(taskLogScheme, "id=%s" % taskLog.id)
116 
117 
Here is the call graph for this function:

Member Data Documentation

◆ dbi

tests.test_dtm_TaskLogScheme.TestTaskLogScheme.dbi

Definition at line 20 of file test_dtm_TaskLogScheme.py.


The documentation for this class was generated from the following file: