Create a valid ISIN

For a suite of tests I recently wrote, I had to create valid ISINs. The official page doesn’t give away much on how the final checksum digit should be computed and the examples on Wikipedia are, in my opinion, not particularly clear.

While I was trying to understand more, I stumbled upon a related unanswered question on stack overflow, which was basically after what I was trying to do, so I took the time to answer it.

Here I report a slightly modified version of the code snippet linked above, to create the checksum digit for the first 11 characters of an ISIN.

import string

def digit_sum(n):
    return (n // 10) + (n % 10)

alphabet = {letter: value for (value, letter) in
            enumerate(''.join(str(n) for n in range(10)) + string.uppercase)}

def isinChecksumDigit(isin):
    isin_to_digits = ''.join(str(d) for d in (alphabet[v] for v in isin))
    isin_sum = 0
    for (i, c) in enumerate(reversed(isin_to_digits), 1):
        if i % 2 == 1:
            isin_sum += digit_sum(2*int(c))
            isin_sum += int(c)

    checksum_digit = abs(- isin_sum % 10)
    return checksum_digit

Assuming countries is a list of valid, i.e. ISO-6166 compliant, country codes, you call the isinChecksumDigit as follows:

In [1]: isin = '{:2s}{:09d}'.format(random.choice(countries), random.randint(1, 10E8))

In [2]: isin
Out[2]: 'KR681111517'

In [3]: validIsin = isin + str(isinChecksumDigit(isin))

In [4]: validIsin
Out[4]: 'KR6811115171

Remote tail

You might have wondered why I felt the urge to specify “local” in the title of last post. Well, fast forward a few days since then, for a similar set of tests I also needed to check a log file on a remote Linux machine – that is, I needed some kind of remote tail.

ssh tail -f

We already know select will be part of our tool set. On top of that, we’ll need to forward the command across the network – and a nice way of doing that is over SSH. In Python, this task is relatively simple if you choose to use paramiko, a third party library implementing the SSHv2 protocol.

A few caveats here, as well. The following snippet is a raw prototype to demonstrate the functionality. It fit my bills, but YMMV. Of course many aspects can be improved, starting from instance with isBeginningOfMessage, which is much better placed in a derived class, so that different BOM patterns can be handled. Closing the SSH channel cleanly is also something you might want to polish before using this class.

import paramiko
import re
import select
import Queue

class SSHTail(object):
    """Tail a remote file and store new messages in a queue
    READ_ONLY = select.POLLIN | select.POLLPRI | select.POLLHUP | select.POLLERR
    TIMEOUT = 1000  # milliseconds
    BUF_SIZE = 1024
    NEWLINE_CHARS = {'\n', '\r'}

    def __init__(self, host, path): = host
        self.path = path
        self.poller = select.poll()
        self.messageQueue = Queue.deque()

    def start(self):
        """Start the tail command and return the queue used to
        store read messages
        client = paramiko.SSHClient()
        self.client = client

        transport = self.client.get_transport()
        self.transport = transport

        channel = self.transport.open_session()
        channel.exec_command("tail -F %s" % self.path) = channel

        self.poller.register(, self.READ_ONLY)

        return self.messageQueue

    def isBeginningOfMessage(line):
        """Return True if the line starts with the hardcoded Beginning of
        Message pattern
        BOMPattern = ''
        return re.match(BOMPattern, line)

    def loop(self):
        """Whilst the SSH tunnel is active, keep polling for new
        content and call parseBuffer() to parse it into messages
        while self.transport.is_active():
            events = self.poller.poll(self.TIMEOUT)
            for fd, flag in events:
                if flag & (select.POLLIN | select.POLLPRI):
                    buf =

    def parseBuffer(self, buf):
        """Given a buffer buf, split it into messages and glue it together to
        previous messages, if buf is not the beginning of a message.
        Note: assumes each message is on its on line.
        if buf:
            messages = buf.splitlines()

            oldest = messages[0]
            if not self.isBeginningOfMessage(oldest):
                    messages[0] = self.messageQueue.popleft() + oldest
                except IndexError:

            for message in messages:

Local tail

tail -f

A while ago I needed, for one of my tests, to monitor a log file on a Linux system and store any new lines, so that I could access the added content at the end of the test. In a sense, I needed a kind of buffered tail -f on a local file.

A quick search led me to the select module.
Without further ado, here’s the code to watch one or more files, and to store anything added to those files in a message queue.

It’s a quick and dirty version which can be improved in many ways. For starters, the keys to access the message queues are the sockets themselves, pretty useless in general, but good enough in my case. Second, notice the file is never closed explicitly: definitely not ideal.

import Queue
import select

class Watcher(object):
    TIMEOUT = 1000
    READ_ONLY = select.POLLIN | select.POLLPRI | select.POLLHUP | select.POLLERR

    def __init__(self):
        """Initialize the Watcher"""
        self.poller = select.poll()

        self.fd_to_socket = {}
        self.message_queues = {}

    def addFile(self, path):
        """Add a file to monitor.
        :path: absolute path of the file, including the filename
        f = open(path)
        self.poller.register(f, self.READ_ONLY)
        self.fd_to_socket[f.fileno()] = f 
        self.message_queues[f] = Queue.deque()

    def start(self):
        """Start polling files"""
        while True:
            events = self.poller.poll(self.TIMEOUT)
            for fd, flag in events:
                s = self.fd_to_socket[fd]
                if flag & (select.POLLIN | select.POLLPRI):
                    lines = s.readlines()
                    if lines:

Nosetests test generators – make your bool values more explicit

I had to write a test generator in Python + nostests to drive the same test in two different conditions.

The way I initially wrote the generator initially resembled something like this:

def testChildIsPopulatedCorrectly(self):
    for case in [True, False]:
        yield self.checkChildIsPopulatedCorrectly, case

Although concise, the test as it’s written is not really easy to follow: there is no way to know what the test does without delving into the checkChildIsPopulatedCorrectly method and see which paths case triggers.

A second problem happens when the test runs. The output will be:

testChildIsPopulatedCorrectly(True,) ... ok
testChildIsPopulatedCorrectly(False,) ... ok

Again, it is not very clear what True and False mean in that context.

A solution is to create two simple classes, within the generator[1]:

def testChildIsPopulatedCorrectly(self):
    class WaitForChildren(object):
        def __bool__(self): return True  # [2]
        def __repr__(self): return 'Wait until all the children in the tree are populated'
    class DontWaitForChildren(object):
        def __bool__(self): return False
        def __repr__(self): return 'Do not wait until all the children in the tree are populated'
    for case in [WaitForChildren(), DontWaitForChildren()]:
        yield self.checkChildIsPopulatedCorrectly, case

The next time you run the tests, you’ll get:

testChildIsPopulatedCorrectly(Wait until all the children in the tree are populated,) ... ok
testChildIsPopulatedCorrectly(Do not wait until all the children in the tree are populated,) ... ok

Now both the test generator and the output are clearer.

[1] This is a proof of concept. The complete solution I implemented involves the definition of a more generic class in a different module.

[2] just a heads-up that you should include __nonzero__ = __bool__ if you are using Python 2.x.

Raising (and catching!) exceptions in separate threads

When coding some system tests, I needed to create a Watchdog class: a context manager that upon entering the context starts a threading.Timer and upon exiting cancels it. If the timer goes off before the context is exited, an exception is thrown.

When writing unit tests for the class, I realised that the Watchdog would raise the exception in its own thread, with the end result that I could not catch it in the main thread.
The test output was something like this:

test_watchdog.TestWatchdog.test_raiseException ... Exception in thread Thread-1 [...]

You can fix that by means of a function in the Python API, PyThreadState_SetAsyncExc (ctypes module), which raises an asynchronous exception in a separate thread.

I later discovered that there is a module that does pretty much the same (with some bonus features such as the definition of a ‘timeoutable’ decorator): it is called stopit. Go get it if you’re into that sort of things!

Powershell in action, parsing XML

Back from the Christmas break, after a very busy period, let’s catch up on Powershell and XML.

Right before the break I was presented with a problem of parsing some XML files. The files all look like this:

<CONTAINER Name="X" Value="Y" Key="Z" ...>
    <NOTE Label="FirstCategory">02941ca7-43f7-4724-98c6-838f46bb2df9</NOTE>
    <NOTE Label="SecondCategory">b44561be-2176-48f7-9aa2-2ec70949cd7f</NOTE>
    <NOTE Label="ThirdCategory">48958fa1-64f7-49f2-a522-63f3ff81af91</NOTE>
    <NOTE Label="FourthCategory">5c826f70-388b-443d-a9b0-e5da21b85584</NOTE>

I needed to grab the value of the attribute Name from the <CONTAINER> element and the inner text of any <NOTE> element for which the attribute Label was set to FourthCategory.

No explanation this time, as the code is really simple. In fact, we’ve already seen what the -join and -eq operators do. Add a couple of foreach et voilà:

foreach ($c in $res.xml.CONTAINER) {
  foreach ($a in $c.NOTES) {
    $el = $a.note | Where { $_.Label -eq "FourthCategory" }
  $c.Name,$el."#text" -join ","

Running Coded-UI automated tests from the command line

Short and concise post, more like a memory aid for myself, about running Coded-UI tests without the need of the Visual Studio GUI. Disclaimer: I’m talking about VS 2012 (though this may apply to VS 2010 too).

According to MSDN,

MSTest is used to run load test in Visual Studio 2012. By default, unit tests and coded UI tests use VSTest.Console.exe in Visual Studio 2012. However, MSTest is used for compatibility with test projects created using Visual Studio 2010, or if a .testsettings file is manually added to a Visual Studio 2012 solution containing a unit test project, or coded UI test project.

Indeed you can run Coded-UI tests from the VS 2012 command prompt, by simply issuing VSTest.Console.exe NameOfYourTestSuite.dll. However, you could do the same by issuing MSTest.exe NameOfYourTestSuite.dll without the need of a .testsettings file.

It goes without saying, beware that if you try running your tests using MSTest from within a VS 2010 prompt, you’ll most likely end up having the command whinging like this: Unable to load the test container 'NameOfYourTestSuite.dll' or one of its dependencies. Error details: System.IO.FileNotFoundException: Could not load file or assembly 'Microsoft.VisualStudio.TestTools.UITesting, Version=, Culture=neutral, PublicKeyToken=b03f5f7f11d50a3a' or one of its dependencies. The system cannot find the file specified.

That said, if you want to be able to run similar tests on a machine that does not have VS 2012 installed, you just need to download the Agents for Visual Studio 2012.