Explorar o código

[sqoop] Improve testing to show real error

Abraham Elmahrek %!s(int64=12) %!d(string=hai) anos
pai
achega
f2a2ab629d
Modificáronse 1 ficheiros con 18 adicións e 4 borrados
  1. 18 4
      apps/sqoop/src/sqoop/tests.py

+ 18 - 4
apps/sqoop/src/sqoop/tests.py

@@ -82,6 +82,16 @@ class TestSqoopServerBase(SqoopServerProvider):
 
 
     return self.client.create_job(job)
     return self.client.create_job(job)
 
 
+  def delete_sqoop_object(self, obj):
+    if isinstance(obj, Connection):
+      self.client.delete_connection(obj)
+    elif isinstance(obj, Job):
+      self.client.delete_job(obj)
+
+  def delete_sqoop_objects(self, objects):
+    for obj in objects:
+      self.delete_sqoop_object(obj)
+
 class TestSqoopClientConnections(TestSqoopServerBase):
 class TestSqoopClientConnections(TestSqoopServerBase):
   def test_connection(self):
   def test_connection(self):
     try:
     try:
@@ -110,10 +120,13 @@ class TestSqoopClientConnections(TestSqoopServerBase):
 
 
 class TestSqoopClientJobs(TestSqoopServerBase):
 class TestSqoopClientJobs(TestSqoopServerBase):
   def test_job(self):
   def test_job(self):
+    removable = []
     try:
     try:
       # Create
       # Create
       conn = self.create_connection(name='conn3')
       conn = self.create_connection(name='conn3')
+      removable.append(conn)
       job = self.create_job("IMPORT", "job1", connection_id=conn.id)
       job = self.create_job("IMPORT", "job1", connection_id=conn.id)
+      removable.insert(0, job)
       assert_true(job.id)
       assert_true(job.id)
 
 
       job2 = self.client.get_job(job.id)
       job2 = self.client.get_job(job.id)
@@ -125,17 +138,18 @@ class TestSqoopClientJobs(TestSqoopServerBase):
       job3 = self.client.update_job(job)
       job3 = self.client.update_job(job)
       assert_equal(job.name, job3.name)
       assert_equal(job.name, job3.name)
     finally:
     finally:
-      self.client.delete_job(job)
-      self.client.delete_connection(conn)
+      self.delete_sqoop_objects(removable)
 
 
   def test_get_jobs(self):
   def test_get_jobs(self):
+    removable = []
     try:
     try:
       conn = self.create_connection(name='conn4')
       conn = self.create_connection(name='conn4')
+      removable.append(conn)
       job = self.create_job("IMPORT", "job2", connection_id=conn.id)
       job = self.create_job("IMPORT", "job2", connection_id=conn.id)
+      removable.insert(0, job)
       assert_true(job.id)
       assert_true(job.id)
 
 
       jobs = self.client.get_jobs()
       jobs = self.client.get_jobs()
       assert_true(len(jobs) > 0)
       assert_true(len(jobs) > 0)
     finally:
     finally:
-      self.client.delete_job(job)
-      self.client.delete_connection(conn)
+      self.delete_sqoop_objects(removable)