Merge lp:~therve/landscape-client/fake-package-reporter into lp:~landscape/landscape-client/trunk

[Thomas Herve]

The branch introduces 2 reporters: one named global or "master", which operates as a normal reporter but stores the messages sent. The other has a pointer to that store, and send those messages instead of generating new ones using smart. I used env variables to instrumentate it, which is not great but simple. Using different plugins means creating new modules for each one, and new scripts.

[Free Ekanayaka (free.ekanayaka)]

Looks good! I have a few comments though.

[1]

+class FakePackageStore(PackageStore):

It'd be good to add direct tests for this class.

[2]

+ if not os.path.exists(global_store):
+ return succeed(None)

Why is this needed?

[3]

+ global_db = sqlite3.connect(global_store)
+ cursor = global_db.cursor()
+ all_message_ids = set(
+ row[0] for row in
+ cursor.execute("SELECT id FROM message").fetchall())

Can't FakePackageStore.get_message_ids be used instead?

[4]

+ messages = list(
+ (row[0], row[1]) for row in
+ cursor.execute(
+ "SELECT id, data FROM message WHERE id IN (%s) "
+ "ORDER BY id" % params, tuple(not_sent)).fetchall())

It'd be nice to encapsulate this if FakePackageStore, and test it separately.

[5]

+ lambda x, message=message:
+ self._broker.send_message(message, True))

Ah, nice trick to get the message variable in the closure.

[6]

+ def send_message(self, message):
+ return succeed(None)
+
+ def use_hash_id_db(self):
+ return succeed(None)
+
+ def request_unknown_hashes(self):
+ return succeed(None)
+
+ def remove_expired_hash_id_requests(self):
+ return succeed(None)
+
+ def handle_task(self, task):
+ return succeed(None)
+

Instead of this, why not override the run() method directly? It would avoid having to modify the fake reporter in case we add methods to the real one.

review: Needs Fixing

[lp:~therve/landscape-client/fake-package-reporter]

364. By Thomas Herve on 2011-08-19

Move SQL in the store

365. By Thomas Herve on 2011-08-19

Implement run instead

[Thomas Herve (therve)]

[1] This is for load testing, so I don't think we need to go out of our way to test this...

[2] The fake clients make start a bit before the master one, so I want to make sure it doesn't crash.

[3] Yep done.

[4] Moved.

[6] Fixed.

Thanks!

[Free Ekanayaka (free.ekanayaka)]

Tahnks, +1!

[7]

+ result.addCallback(lambda x: self.handle_tasks())

Maybe this can just be

         result.addCallback(lambda x: self._store.clear_tasks())

and handle_task() can be dropped.

[8]

+ def __init__(self, package_store, package_facade, remote_broker, config):
+ super(FakeGlobalReporter, self).__init__(package_store, package_facade,
+ remote_broker, config)

Do we need this?

review: Approve

[Frank Wierzbicki (fwierzbicki)]

I can't add anything to Free's comments - looks good to me +1!

review: Approve

=== modified file 'landscape/broker/store.py'
--- landscape/broker/store.py	2011-07-21 23:55:47 +0000
+++ landscape/broker/store.py	2011-08-19 14:11:42 +0000
@@ -43,9 +43,7 @@
     api = SERVER_API
 
     def __init__(self, persist, directory, directory_size=1000,
-                 monitor_interval=60*60, get_time=time.time):
-        """
-        """
+                 monitor_interval=60 * 60, get_time=time.time):
         self._get_time = get_time
         self._directory = directory
         self._directory_size = directory_size
@@ -82,8 +80,8 @@
     def get_sequence(self):
         """Get the current sequence.
 
-        @return: The sequence number of the message that the server expects us to
-           send on the next exchange.
+        @return: The sequence number of the message that the server expects us
+            to send on the next exchange.
         """
         return self._persist.get("sequence", 0)
 
@@ -98,8 +96,8 @@
     def get_server_sequence(self):
         """Get the current server sequence.
 
-        @return: the sequence number of the message that we will ask the server to
-            send to us on the next exchange.
+        @return: the sequence number of the message that we will ask the server
+            to send to us on the next exchange.
         """
         return self._persist.get("server_sequence", 0)
 
@@ -161,7 +159,7 @@
 
     def delete_old_messages(self):
         """Delete messages which are unlikely to be needed in the future."""
-        for fn in itertools.islice(self._walk_messages(exclude=HELD+BROKEN),
+        for fn in itertools.islice(self._walk_messages(exclude=HELD + BROKEN),
                                    self.get_pending_offset()):
             os.unlink(fn)
             containing_dir = os.path.split(fn)[0]
@@ -256,7 +254,8 @@
     def _walk_pending_messages(self):
         """Walk the files which are definitely pending."""
         pending_offset = self.get_pending_offset()
-        for i, filename in enumerate(self._walk_messages(exclude=HELD+BROKEN)):
+        for i, filename in enumerate(self._walk_messages(exclude=HELD +
+                                                                 BROKEN)):
             if i >= pending_offset:
                 yield filename
 
@@ -308,10 +307,10 @@
                     if accepted:
                         new_filename = self._get_next_message_filename()
                         os.rename(old_filename, new_filename)
-                        self._set_flags(new_filename, set(flags)-set(HELD))
+                        self._set_flags(new_filename, set(flags) - set(HELD))
                 else:
                     if not accepted and offset >= pending_offset:
-                        self._set_flags(old_filename, set(flags)|set(HELD))
+                        self._set_flags(old_filename, set(flags) | set(HELD))
                     offset += 1
 
     def _get_flags(self, path):
@@ -324,16 +323,18 @@
         dirname, basename = os.path.split(path)
         new_path = os.path.join(dirname, basename.split("_")[0])
         if flags:
-            new_path += "_"+"".join(sorted(set(flags)))
+            new_path += "_" + "".join(sorted(set(flags)))
         os.rename(path, new_path)
         return new_path
 
     def _add_flags(self, path, flags):
-        self._set_flags(path, self._get_flags(path)+flags)
+        self._set_flags(path, self._get_flags(path) + flags)
 
 
 def get_default_message_store(*args, **kwargs):
-    """Get a L{MessageStore} object with all Landscape message schemas added."""
+    """
+    Get a L{MessageStore} object with all Landscape message schemas added.
+    """
     from landscape. message_schemas import message_schemas
     store = MessageStore(*args, **kwargs)
     for schema in message_schemas.values():

=== modified file 'landscape/package/reporter.py'
--- landscape/package/reporter.py	2011-08-10 14:56:43 +0000
+++ landscape/package/reporter.py	2011-08-19 14:11:42 +0000
@@ -10,10 +10,11 @@
 from landscape.lib.twisted_util import gather_results, spawn_process
 from landscape.lib.fetch import fetch_async
 from landscape.lib.fs import touch_file
+from landscape.lib import bpickle
 
 from landscape.package.taskhandler import (
     PackageTaskHandlerConfiguration, PackageTaskHandler, run_task_handler)
-from landscape.package.store import UnknownHashIDRequest
+from landscape.package.store import UnknownHashIDRequest, FakePackageStore
 
 
 HASH_ID_REQUEST_TIMEOUT = 7200
@@ -79,6 +80,9 @@
         result.callback(None)
         return result
 
+    def send_message(self, message):
+        return self._broker.send_message(message, True)
+
     def fetch_hash_id_db(self):
         """
         Fetch the appropriate pre-canned database of hash=>id mappings
@@ -219,7 +223,7 @@
             "type": "package-reporter-result",
             "code": code,
             "err": err}
-        return self._broker.send_message(message, True)
+        return self.send_message(message)
 
     def handle_task(self, task):
         message = task.data
@@ -389,7 +393,7 @@
         """Create a hash_id_request and send message with "request-id"."""
         request = self._store.add_hash_id_request(unknown_hashes)
         message["request-id"] = request.id
-        result = self._broker.send_message(message, True)
+        result = self.send_message(message)
 
         def set_message_id(message_id):
             request.message_id = message_id
@@ -530,7 +534,7 @@
             return succeed(False)
 
         message["type"] = "packages"
-        result = self._broker.send_message(message, True)
+        result = self.send_message(message)
 
         logging.info("Queuing message with changes in known packages: "
                      "%d installed, %d available, %d available upgrades, "
@@ -595,7 +599,7 @@
             return succeed(False)
 
         message["type"] = "package-locks"
-        result = self._broker.send_message(message, True)
+        result = self.send_message(message)
 
         logging.info("Queuing message with changes in known package locks:"
                      " %d created, %d deleted." %
@@ -613,8 +617,81 @@
         return result
 
 
+class FakeGlobalReporter(PackageReporter):
+    """
+    A standard reporter, which additionally stores messages sent into its
+    package store.
+    """
+
+    package_store_class = FakePackageStore
+
+    def __init__(self, package_store, package_facade, remote_broker, config):
+        super(FakeGlobalReporter, self).__init__(package_store, package_facade,
+                                                 remote_broker, config)
+
+    def send_message(self, message):
+        self._store.save_message(message)
+        return super(FakeGlobalReporter, self).send_message(message)
+
+
+class FakeReporter(PackageReporter):
+    """
+    A fake reporter which only sends messages previously stored by a
+    L{FakeGlobalReporter}.
+    """
+
+    package_store_class = FakePackageStore
+
+    def run(self):
+        result = succeed(None)
+
+        # If the appropriate hash=>id db is not there, fetch it
+        result.addCallback(lambda x: self.fetch_hash_id_db())
+
+        result.addCallback(lambda x: self.handle_tasks())
+
+        # Finally, verify if we have anything new to send to the server.
+        result.addCallback(lambda x: self.send_pending_messages())
+
+        return result
+
+    def handle_task(self, task):
+        return succeed(None)
+
+    def send_pending_messages(self):
+        """
+        As the last callback of L{PackageReporter}, sends messages stored.
+        """
+        global_file = os.environ["FAKE_PACKAGE_STORE"]
+        if not os.path.exists(global_file):
+            return succeed(None)
+        message_sent = set(self._store.get_message_ids())
+        global_store = FakePackageStore(global_file)
+        all_message_ids = set(global_store.get_message_ids())
+        not_sent = all_message_ids - message_sent
+        deferred = succeed(None)
+        got_type = set()
+        if not_sent:
+            messages = global_store.get_messages_by_ids(not_sent)
+            sent = []
+            for message_id, message in messages:
+                message = bpickle.loads(str(message))
+                if message["type"] not in got_type:
+                    got_type.add(message["type"])
+                    sent.append(message_id)
+                    deferred.addCallback(
+                        lambda x, message=message: self.send_message(message))
+            self._store.save_message_ids(sent)
+        return deferred
+
+
 def main(args):
-    return run_task_handler(PackageReporter, args)
+    if "FAKE_PACKAGE_STORE" in os.environ:
+        return run_task_handler(FakeReporter, args)
+    elif "FAKE_GLOBAL_PACKAGE_STORE" in os.environ:
+        return run_task_handler(FakeGlobalReporter, args)
+    else:
+        return run_task_handler(PackageReporter, args)
 
 
 def find_reporter_command():

=== modified file 'landscape/package/store.py'
--- landscape/package/store.py	2011-07-21 23:55:47 +0000
+++ landscape/package/store.py	2011-08-19 14:11:42 +0000
@@ -329,6 +329,40 @@
                        ",".join([str(task.id) for task in except_tasks]))
 
 
+class FakePackageStore(PackageStore):
+    """
+    A L{PackageStore} with an additional message table to store sent messages.
+    """
+
+    def _ensure_schema(self):
+        super(FakePackageStore, self)._ensure_schema()
+        ensure_fake_package_schema(self._db)
+
+    @with_cursor
+    def save_message(self, cursor, message):
+        cursor.execute("INSERT INTO message (data) VALUES (?)",
+                       (buffer(bpickle.dumps(message)),))
+
+    @with_cursor
+    def get_message_ids(self, cursor):
+        return [row[0] for row in
+                cursor.execute("SELECT id FROM message").fetchall()]
+
+    @with_cursor
+    def save_message_ids(self, cursor, message_ids):
+        cursor.executemany(
+            "INSERT INTO message (id) VALUES (?)",
+            [(message_id,) for message_id in message_ids])
+
+    @with_cursor
+    def get_messages_by_ids(self, cursor, message_ids):
+        params = ", ".join(["?"] * len(message_ids))
+        result = cursor.execute(
+                "SELECT id, data FROM message WHERE id IN (%s) "
+                "ORDER BY id" % params, tuple(message_ids)).fetchall()
+        return [(row[0], row[1]) for row in result]
+
+
 class HashIDRequest(object):
 
     def __init__(self, db, id):
@@ -447,3 +481,16 @@
     else:
         cursor.close()
         db.commit()
+
+
+def ensure_fake_package_schema(db):
+    cursor = db.cursor()
+    try:
+        cursor.execute("CREATE TABLE message"
+                       " (id INTEGER PRIMARY KEY, data BLOB)")
+    except (sqlite3.OperationalError, sqlite3.DatabaseError):
+        cursor.close()
+        db.rollback()
+    else:
+        cursor.close()
+        db.commit()

=== modified file 'landscape/package/taskhandler.py'
--- landscape/package/taskhandler.py	2011-06-13 23:58:09 +0000
+++ landscape/package/taskhandler.py	2011-08-19 14:11:42 +0000
@@ -85,6 +85,7 @@
 
     queue_name = "default"
     lsb_release_filename = LSB_RELEASE_FILENAME
+    package_store_class = PackageStore
 
     def __init__(self, package_store, package_facade, remote_broker, config):
         self._store = package_store
@@ -261,7 +262,7 @@
     # import Smart unless we need to.
     from landscape.package.facade import SmartFacade
 
-    package_store = PackageStore(config.store_filename)
+    package_store = cls.package_store_class(config.store_filename)
     package_facade = SmartFacade()
 
     def finish():

=== modified file 'landscape/package/tests/test_reporter.py'
--- landscape/package/tests/test_reporter.py	2011-08-10 14:56:43 +0000
+++ landscape/package/tests/test_reporter.py	2011-08-19 14:11:42 +0000
@@ -8,15 +8,18 @@
 from twisted.internet import reactor
 
 from landscape.lib.fetch import fetch_async, FetchError
-from landscape.package.store import PackageStore, UnknownHashIDRequest
+from landscape.lib import bpickle
+from landscape.package.store import (
+    PackageStore, UnknownHashIDRequest, FakePackageStore)
 from landscape.package.reporter import (
     PackageReporter, HASH_ID_REQUEST_TIMEOUT, main, find_reporter_command,
-    PackageReporterConfiguration)
+    PackageReporterConfiguration, FakeGlobalReporter, FakeReporter)
 from landscape.package import reporter
 from landscape.package.facade import SmartFacade
 from landscape.package.tests.helpers import (
     SmartFacadeHelper, HASH1, HASH2, HASH3)
-from landscape.tests.helpers import LandscapeTest, BrokerServiceHelper
+from landscape.tests.helpers import (
+    LandscapeTest, BrokerServiceHelper, EnvironSaverHelper)
 from landscape.tests.mocker import ANY
 
 SAMPLE_LSB_RELEASE = "DISTRIB_CODENAME=codename\n"
@@ -1589,6 +1592,130 @@
         return deferred
 
 
+class GlobalPackageReporterTest(LandscapeTest):
+
+    helpers = [SmartFacadeHelper, BrokerServiceHelper]
+
+    def setUp(self):
+
+        def set_up(ignored):
+            self.store = FakePackageStore(self.makeFile())
+            self.config = PackageReporterConfiguration()
+            self.reporter = FakeGlobalReporter(
+                self.store, self.facade, self.remote, self.config)
+            self.config.data_path = self.makeDir()
+            os.mkdir(self.config.package_directory)
+
+        result = super(GlobalPackageReporterTest, self).setUp()
+        return result.addCallback(set_up)
+
+    def test_store_messages(self):
+        """
+        L{FakeGlobalReporter} stores messages which are sent.
+        """
+        message_store = self.broker_service.message_store
+        message_store.set_accepted_types(["package-reporter-result"])
+        self.reporter.smart_update_filename = self.makeFile(
+            "#!/bin/sh\necho -n error >&2\necho -n output\nexit 0")
+        os.chmod(self.reporter.smart_update_filename, 0755)
+        deferred = Deferred()
+
+        def do_test():
+            result = self.reporter.run_smart_update()
+
+            def callback(ignore):
+                message = {"type": "package-reporter-result",
+                           "code": 0, "err": u"error"}
+                self.assertMessages(
+                    message_store.get_pending_messages(), [message])
+                stored = list(self.store._db.execute(
+                    "SELECT id, data FROM message").fetchall())
+                self.assertEqual(1, len(stored))
+                self.assertEqual(1, stored[0][0])
+                self.assertEqual(message, bpickle.loads(str(stored[0][1])))
+            result.addCallback(callback)
+            result.chainDeferred(deferred)
+
+        reactor.callWhenRunning(do_test)
+        return deferred
+
+
+class FakePackageReporterTest(LandscapeTest):
+
+    helpers = [EnvironSaverHelper, SmartFacadeHelper, BrokerServiceHelper]
+
+    def setUp(self):
+
+        def set_up(ignored):
+            self.store = FakePackageStore(self.makeFile())
+            global_file = self.makeFile()
+            self.global_store = FakePackageStore(global_file)
+            os.environ["FAKE_PACKAGE_STORE"] = global_file
+            self.config = PackageReporterConfiguration()
+            self.reporter = FakeReporter(
+                self.store, self.facade, self.remote, self.config)
+            self.config.data_path = self.makeDir()
+            os.mkdir(self.config.package_directory)
+
+        result = super(FakePackageReporterTest, self).setUp()
+        return result.addCallback(set_up)
+
+    def test_send_messages(self):
+        """
+        L{FakeReporter} sends messages stored in the global store specified by
+        C{FAKE_PACKAGE_STORE}.
+        """
+        message_store = self.broker_service.message_store
+        message_store.set_accepted_types(["package-reporter-result"])
+        message = {"type": "package-reporter-result",
+                   "code": 0, "err": u"error"}
+        self.global_store.save_message(message)
+
+        def check(ignore):
+            self.assertMessages(
+                message_store.get_pending_messages(), [message])
+            stored = list(self.store._db.execute(
+                "SELECT id FROM message").fetchall())
+            self.assertEqual(1, len(stored))
+            self.assertEqual(1, stored[0][0])
+
+        deferred = self.reporter.run()
+        deferred.addCallback(check)
+        return deferred
+
+    def test_filter_message_type(self):
+        """
+        L{FakeReporter} only sends one message of each type per run.
+        """
+        message_store = self.broker_service.message_store
+        message_store.set_accepted_types(["package-reporter-result"])
+        message1 = {"type": "package-reporter-result",
+                    "code": 0, "err": u"error"}
+        self.global_store.save_message(message1)
+        message2 = {"type": "package-reporter-result",
+                    "code": 1, "err": u"error"}
+        self.global_store.save_message(message2)
+
+        def check1(ignore):
+            self.assertMessages(
+                message_store.get_pending_messages(), [message1])
+            stored = list(self.store._db.execute(
+                "SELECT id FROM message").fetchall())
+            self.assertEqual(1, stored[0][0])
+            return self.reporter.run().addCallback(check2)
+
+        def check2(ignore):
+            self.assertMessages(
+                message_store.get_pending_messages(), [message1, message2])
+            stored = list(self.store._db.execute(
+                "SELECT id FROM message").fetchall())
+            self.assertEqual(2, len(stored))
+            self.assertEqual(1, stored[0][0])
+            self.assertEqual(2, stored[1][0])
+
+        return self.reporter.run().addCallback(check1)
+
+
 class EqualsHashes(object):
 
     def __init__(self, *hashes):