Changeset 12094

Timestamp:

30 Nov 2014, 08:12:27 (10 years ago)

Author:

Henrik Bettermann

Message:

Add applications_catalog. We need this catalog for imports.

Add catalog tests.

Add attribute last_product_id. This ensures that the last product can be identified even if the product assigned to the application has been removed.

Location:

main/waeup.ikoba/trunk/src/waeup/ikoba/customers

Files:

• applications.py (modified) (1 diff)
• batching.py (modified) (2 diffs)
• browser.py (modified) (4 diffs)
• catalog.py (modified) (2 diffs)
• interfaces.py (modified) (2 diffs)
• tests/test_browser.py (modified) (3 diffs)
• tests/test_catalog.py (modified) (5 diffs)

main/waeup.ikoba/trunk/src/waeup/ikoba/customers/applications.py

@@ -72 +72 @@
         except AttributeError:
             self.application_id = u'a123'
+        self.last_product_id = None
         return
 

main/waeup.ikoba/trunk/src/waeup/ikoba/customers/batching.py

@@ -43 +43 @@
 from waeup.ikoba.customers.interfaces import (
     ICustomer, ICustomerUpdateByRegNo,
-    ICustomerDocument)
+    ICustomerDocument, IApplication)
 from waeup.ikoba.customers.workflow import  (
     IMPORTABLE_REGISTRATION_STATES, IMPORTABLE_REGISTRATION_TRANSITIONS,
@@ -461 +461 @@
                 errs.append(('document_id','invalid format'))
         return errs, inv_errs, conv_dict
+
+
+class ApplicationProcessor(CustomerProcessorBase):
+    """A batch processor for IApplication objects.
+    """
+    grok.implements(IBatchProcessor)
+    grok.provides(IBatchProcessor)
+    grok.context(Interface)
+    util_name = 'applicationprocessor'
+    grok.name(util_name)
+
+    name = _('Application Processor')
+    iface = IApplication
+    factory_name = 'waeup.SampleApplication'
+
+    location_fields = []
+    additional_fields = ['application_id', 'class_name']
+    additional_headers = ['class_name']
+
+    def checkHeaders(self, headerfields, mode='ignore'):
+        super(ApplicationProcessor, self).checkHeaders(headerfields)
+        if mode in ('update', 'remove') and not 'application_id' in headerfields:
+            raise FatalCSVError(
+                "Need application_id for import in update and remove modes!")
+        return True
+
+    def getParent(self, row, site):
+        customer = self._getCustomer(row, site)
+        if customer is None:
+            return None
+        return customer['applications']
+
+    def getEntry(self, row, site):
+        applications = self.getParent(row, site)
+        if applications is None:
+            return None
+        application_id = row.get('application_id', None)
+        if application_id is None:
+            return None
+        entry = applications.get(application_id)
+        return entry
+
+    def updateEntry(self, obj, row, site, filename):
+        """Update obj to the values given in row.
+        """
+        items_changed = super(ApplicationProcessor, self).updateEntry(
+            obj, row, site, filename)
+        customer = self.getParent(row, site).__parent__
+        customer.__parent__.logger.info(
+            '%s - %s - %s - updated: %s'
+            % (self.name, filename, customer.customer_id, items_changed))
+        return
+
+    def addEntry(self, obj, row, site):
+        parent = self.getParent(row, site)
+        application_id = row['application_id'].strip('#')
+        parent[application_id] = obj
+        return
+
+    def delEntry(self, row, site):
+        application = self.getEntry(row, site)
+        parent = self.getParent(row, site)
+        if application is not None:
+            customer = self._getCustomer(row, site)
+            customer.__parent__.logger.info('%s - Application removed: %s'
+                % (customer.customer_id, application.application_id))
+            del parent[application.application_id]
+        return
+
+    def checkConversion(self, row, mode='ignore'):
+        """Validates all values in row.
+        """
+        errs, inv_errs, conv_dict = super(
+            ApplicationProcessor, self).checkConversion(row, mode=mode)
+        # We need to check if the class_name corresponds with the
+        # processor chosen. This is to avoid accidentally wrong imports.
+        if mode != 'remove':
+            class_name = row.get('class_name', None)
+            if class_name != self.factory_name.strip('waeup.'):
+                errs.append(('class_name','wrong processor'))
+        # We have to check application_id.
+        application_id = row.get('application_id', None)
+        if mode == 'create':
+            if not application_id:
+                application_id = generate_application_id()
+                conv_dict['application_id'] = application_id
+                return errs, inv_errs, conv_dict
+            cat = queryUtility(ICatalog, name='applications_catalog')
+            results = list(
+                cat.searchResults(application_id=(application_id, application_id)))
+            if results:
+                # application_id must not exist.
+                errs.append(('application_id','id exists'))
+        else:
+            if not application_id.startswith('a'):
+                errs.append(('application_id','invalid format'))
+        return errs, inv_errs, conv_dict

main/waeup.ikoba/trunk/src/waeup/ikoba/customers/browser.py

@@ -67 +67 @@
     if changed_fields:
         changed_fields = reduce(lambda x,y: x+y, changed_fields.values())
+    if 'product' in changed_fields:
+        view.context.last_product_id = data['product'].product_id
     fields_string = ' + '.join(changed_fields)
     view.flash(_('Form has been saved.'))
@@ -1150 +1152 @@
     pnav = 4
 
-    #@property
-    #def label(self):
-    #    return _('${a}: Application ${b}', mapping = {
-    #        'a':self.context.customer.display_fullname,
-    #        'b':self.context.application_id})
-
     @property
     def label(self):
@@ -1172 +1168 @@
     deletion_warning = _('Are you sure?')
 
-    #@property
-    #def label(self):
-    #    return _('${a}: Application ${b}', mapping = {
-    #        'a':self.context.customer.display_fullname,
-    #        'b':self.context.application_id})
-
     @property
     def label(self):
@@ -1205 +1195 @@
         return
 
-    @action(_('Final Submit'), warning=WARNING)
+    @action(_('Apply now (final submit)'), warning=WARNING)
     def finalsubmit(self, **data):
         msave(self, **data)

main/waeup.ikoba/trunk/src/waeup/ikoba/customers/catalog.py

@@ -27 +27 @@
 from waeup.ikoba.interfaces import (
     ICompany, IQueryResultItem)
-from waeup.ikoba.customers.interfaces import ICustomer
+from waeup.ikoba.customers.interfaces import ICustomer, IApplication
 
 
@@ -117 +117 @@
     cat_name = 'customers_catalog'
     defaults = dict(customer_id=None)  # make sure we get all studs by default
+
+
+# Catalog for customer applications
+
+class ApplicationIndexes(grok.Indexes):
+    """A catalog for all applications.
+    """
+    grok.site(ICompany)
+    grok.name('applications_catalog')
+    grok.context(IApplication)
+
+    application_id = grok.index.Field(attribute='application_id')
+    last_product_id = grok.index.Field(attribute='last_product_id')
+    application_category = grok.index.Field(attribute='application_category')

main/waeup.ikoba/trunk/src/waeup/ikoba/customers/interfaces.py

@@ -273 +273 @@
         'Last transition formatted date string')
     application_category = Attribute('Category for the grouping of products')
+    last_product_id = Attribute(
+        'Id of product recently stored with the application')
 
     title = schema.TextLine(
@@ -288 +290 @@
         title = _(u'Product'),
         source = AppCatProductSource(),
-        required = False,
-        )
+        required = True,
+        )

main/waeup.ikoba/trunk/src/waeup/ikoba/customers/tests/test_browser.py

@@ -1073 +1073 @@
         self.browser.getLink("a102").click()
         self.browser.getLink("Manage").click()
+        self.browser.getControl(name="form.product").value = ['SAM']
         self.browser.getControl(name="form.title").value = 'My second app'
         self.browser.getControl("Save").click()
@@ -1144 +1145 @@
         # ... but NOTSAM not.
         self.assertFalse('<option value="LIC">' in self.browser.contents)
+        # So far last_product_id is None.
+        self.assertTrue(self.customer['applications']['a102'].last_product_id is None)
         self.browser.getControl(name="form.product").value = ['SAM']
         self.browser.getControl("Save").click()
+        # After saving the form, last_product_id is set.
+        self.assertEqual(self.customer['applications']['a102'].last_product_id, 'SAM')
+        self.assertTrue('Form has been saved.' in self.browser.contents)
+        # Saving the form again does not unset last_product_id
         self.assertEqual(application.title, 'My second app')
+        self.browser.getControl("Save").click()
+        self.assertEqual(self.customer['applications']['a102'].last_product_id, 'SAM')
         self.assertTrue('Form has been saved.' in self.browser.contents)
         self.browser.getLink("View").click()
@@ -1153 +1162 @@
         self.browser.getLink("Edit").click()
         self.browser.getControl(name="form.title").value = 'My third app'
-        self.browser.getControl("Final Submit").click()
+        self.browser.getControl("Apply now").click()
         self.assertEqual(application.title, 'My third app')
         self.assertEqual(application.state, 'submitted')

main/waeup.ikoba/trunk/src/waeup/ikoba/customers/tests/test_catalog.py

@@ -1 @@
-## $Id: test_catalog.py 10552 2013-08-28 14:33:33Z henrik $
+## $Id$
 ##
 ## Copyright (C) 2014 Uli Fouquet & Henrik Bettermann
@@ -26 +26 @@
 from waeup.ikoba.testing import FunctionalLayer, FunctionalTestCase
 from waeup.ikoba.customers.customer import Customer
+from waeup.ikoba.customers.applications import SampleApplication
+
 
 class CatalogTestSetup(FunctionalTestCase):
@@ -54 +56 @@
         self.customer_id = customer.customer_id
         self.customer = self.app['customers'][self.customer_id]
-        # Update the customers_catalog
-        notify(grok.ObjectModifiedEvent(self.customer))
+
+        product = createObject('waeup.Product')
+        product.product_id = u'SAM'
+        product.title = u'Our Samle Product'
+        product.application_category = u'sample'
+        self.app['products'].addProduct(product)
+        self.product_id = product.product_id
+        self.product = self.app['products'][self.product_id]
+
+        application = createObject('waeup.SampleApplication')
+        application.title = u'My Samle Application'
+        application.product = product
+        application.last_product_id = product.product_id
+        self.customer['applications'].addApplication(application)
+        self.application_id = application.application_id
+        self.application = self.customer['applications'][self.application_id]
+
+        document = createObject('waeup.CustomerSampleDocument')
+        document.title = u'My Samle Document'
+        self.customer['documents'].addDocument(document)
+        self.document_id = document.document_id
+        self.document = self.customer['documents'][self.document_id]
+
         return
 
@@ -63 +86 @@
         return
 
-class CustomerCatalogTests(CatalogTestSetup):
+
+class CustomersCatalogTests(CatalogTestSetup):
 
     layer = FunctionalLayer
@@ -88 +112 @@
         assert len(results) == 1
         assert results[0] is self.app['customers'][self.customer_id]
+
+
+class DocumentsCatalogTests(CatalogTestSetup):
+
+    layer = FunctionalLayer
+
+    def test_get_catalog(self):
+        # We can get an customers catalog if we wish
+        cat = queryUtility(ICatalog, name='documents_catalog')
+        assert cat is not None
+
+    def test_search_by_id(self):
+        # We can find a certain application id
+        cat = queryUtility(ICatalog, name='documents_catalog')
+        results = cat.searchResults(document_id=(self.document_id,
+                                                self.document_id))
+        results = [x for x in results] # Turn results generator into list
+        assert len(results) == 1
+        assert results[0] is self.customer['documents'][self.document_id]
+
+
+class ApplicationsCatalogTests(CatalogTestSetup):
+
+    layer = FunctionalLayer
+
+    def test_get_catalog(self):
+        # We can get an customers catalog if we wish
+        cat = queryUtility(ICatalog, name='applications_catalog')
+        assert cat is not None
+
+    def test_search_by_id(self):
+        # We can find a certain application id
+        cat = queryUtility(ICatalog, name='applications_catalog')
+        results = cat.searchResults(application_id=(self.application_id,
+                                                self.application_id))
+        results = [x for x in results] # Turn results generator into list
+        assert len(results) == 1
+        assert results[0] is self.customer['applications'][self.application_id]
+
+    def test_search_by_category(self):
+        # We can find a certain name
+        cat = queryUtility(ICatalog, name='applications_catalog')
+        results = cat.searchResults(application_category=('sample', 'sample'))
+        results = [x for x in results] # Turn results generator into list
+        assert len(results) == 1
+        assert results[0] is self.customer['applications'][self.application_id]
+
+    def test_search_by_last_product_id(self):
+        # We can find a certain name
+        cat = queryUtility(ICatalog, name='applications_catalog')
+        results = cat.searchResults(last_product_id=('SAM', 'SAM'))
+        results = [x for x in results] # Turn results generator into list
+        assert len(results) == 1
+        assert results[0] is self.customer['applications'][self.application_id]
+        assert results[0].product is self.product