Generate XML feed from person data

carbon/__init__.py

@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
 from __future__ import absolute_import
 
-from carbon.app import people
+from carbon.app import people, PersonFeed
 from carbon.db import engine, session

carbon/app.py

@@ -1,6 +1,8 @@
 # -*- coding: utf-8 -*-
 from __future__ import absolute_import
+from functools import partial
 
+from lxml import etree as ET
 from sqlalchemy.sql import select
 
 from carbon.db import persons, session
@@ -15,3 +17,32 @@ def people():
     with session() as conn:
         for row in conn.execute(sql):
             yield dict(zip(row.keys(), row))
+
+
+def _ns(namespace, element):
+    return ET.QName(namespace, element)
+
+
+SYMPLECTIC_NS = 'http://www.symplectic.co.uk/hrimporter'
+NSMAP = {None: SYMPLECTIC_NS}
+ns = partial(_ns, SYMPLECTIC_NS)
+
+
+def add_child(parent, element, text, **kwargs):
+    """Add a subelement with text."""
+    child = ET.SubElement(parent, ns(element), nsmap=NSMAP, attrib=kwargs)
+    child.text = text
+    return child
+
+
+class PersonFeed(object):
+    def __init__(self):
+        self._root = ET.Element(ns('records'), nsmap=NSMAP)
+
+    def add(self, person):
+        record = ET.SubElement(self._root, ns('record'), nsmap=NSMAP)
+        add_child(record, 'field', person['MIT_ID'], name='[Proprietary_ID]')
+        add_child(record, 'field', person['KRB_NAME'], name='[Username]')
+
+    def bytes(self):
+        return ET.tostring(self._root, encoding="UTF-8")

carbon/cli.py

@@ -3,7 +3,7 @@
 
 import click
 
-from carbon import engine, people
+from carbon import engine, people, PersonFeed
 
 
 @click.group()
@@ -15,5 +15,7 @@ def main():
 @click.option('--db', default='sqlite:///carbon.db')
 def load(db):
     engine.configure(db)
+    feed = PersonFeed()
     for person in people():
-        click.echo(person)
+        feed.add(person)
+    click.echo(feed.bytes())

requirements.txt

@@ -1,3 +1,4 @@
 click==5.1
+lxml==3.5.0
 SQLAlchemy==1.0.9
 wheel==0.24.0

tests/conftest.py

@@ -2,6 +2,7 @@
 from __future__ import absolute_import
 import os
 
+from lxml.builder import ElementMaker
 import pytest
 
 from carbon.db import engine, session, metadata, persons
@@ -23,11 +24,16 @@ def app_init():
 def load_data():
     with session() as s:
         s.execute(persons.delete())
-    with session() as s:
         s.execute(persons.insert(), [
             {'MIT_ID': '123456', 'KRB_NAME': 'foobar'},
             {'MIT_ID': '098754', 'KRB_NAME': 'foobaz'}
         ])
     yield
     with session() as s:
         s.execute(persons.delete())
+
+
+@pytest.fixture
+def E():
+    return ElementMaker(namespace='http://www.symplectic.co.uk/hrimporter',
+                        nsmap={None: 'http://www.symplectic.co.uk/hrimporter'})

tests/test_app.py

@@ -1,9 +1,11 @@
 # -*- coding: utf-8 -*-
 from __future__ import absolute_import
 
+from lxml import etree as ET
 import pytest
 
 from carbon import people
+from carbon.app import PersonFeed, ns, NSMAP, add_child
 
 
 pytestmark = pytest.mark.usefixtures('load_data')
@@ -15,3 +17,29 @@ def test_people_generates_people():
     assert person['KRB_NAME'] == 'foobar'
     person = next(peeps)
     assert person['KRB_NAME'] == 'foobaz'
+
+
+def test_add_child_adds_child_element(E):
+    xml = E.records(
+        E.record('foobar', {'baz': 'bazbar'})
+    )
+    e = ET.Element(ns('records'), nsmap=NSMAP)
+    add_child(e, 'record', 'foobar', baz='bazbar')
+    assert ET.tostring(e) == ET.tostring(xml)
+
+
+def test_person_feed_uses_namespace():
+    p = PersonFeed()
+    assert p._root.tag == "{http://www.symplectic.co.uk/hrimporter}records"
+
+
+def test_person_feed_adds_person(E):
+    xml = E.records(
+        E.record(
+            E.field('1234', {'name': '[Proprietary_ID]'}),
+            E.field('foobar', {'name': '[Username]'})
+        )
+    )
+    p = PersonFeed()
+    p.add({'MIT_ID': '1234', 'KRB_NAME': 'foobar'})
+    assert p.bytes() == ET.tostring(xml)

tests/test_cli.py

@@ -2,6 +2,7 @@
 from __future__ import absolute_import
 
 from click.testing import CliRunner
+from lxml import etree as ET
 import pytest
 
 from carbon.cli import main
@@ -15,8 +16,18 @@ def runner():
     return CliRunner()
 
 
-def test_load_returns_people(runner):
+def test_load_returns_people(runner, E):
+    xml = E.records(
+        E.record(
+            E.field('123456', name='[Proprietary_ID]'),
+            E.field('foobar', name='[Username]')
+        ),
+        E.record(
+            E.field('098754', name='[Proprietary_ID]'),
+            E.field('foobaz', name='[Username]')
+        )
+    )
     res = runner.invoke(main, ['load', '--db',
                         'sqlite:///tests/db/test.db'])
     assert res.exit_code == 0
-    assert 'foobar' in res.output
+    assert res.output.encode('utf-8') == ET.tostring(xml) + b'\n'