Posts tagged python

How to make the simplest unittests in Python

0

Testing your code is nearly a requirement (even more so in Ruby). Unittests are now the most vital elements for evaluating the quality/viability of a project.
I was a little jealous of Ruby where you don’t have so much to write to implement unittests. Here is a simple example:

1
2
3
4
5
6
7
8
9
require "mymodule"
require "test/unit"
 
class TestMyModule < Test::Unit::TestCase
 
  def test_simple
     assert_equal(1, 1 )
  end
end

Now, using Nose, you can get even shorter code. If you do standard Python projects, you’ll use a setup.py file. To use nose, you do not even need to specify the path where to find the tests, just add two lines (tests_require and test_suite) to call nosetest:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
from setuptools import setup, find_packages
import sys, os
import mymodule
 
version = mymodule.__version__
 
setup(name='myproject',
      version=version,
      description="Module to display blah blah blah.",
      long_description=""" """,
      classifiers=[], # Get strings from http://pypi.python.org/pypi?%3Aaction=list_classifiers
      keywords='mymodule foobar',
      author='Luc Stepniewski',
      author_email='[email protected]',
      url='',
      license='GPL',
      packages=find_packages(exclude=['ez_setup', 'examples', 'tests']),
      include_package_data=True,
      tests_require='nose',
      test_suite='nose.collector',
      zip_safe=False,
      install_requires=[
          # -*- Extra requirements: -*-
          'simplejson',
      ],
      entry_points="""
      # -*- Entry points: -*-
      [console_scripts]
      mymodule = mymodule.mainmodule:main
      """,
      )

Now, to add tests, you just have to create a directory named tests (in the root of your project, where your setup.py resides, and then add a python file()s. No need to add a __init__.py to set the directory as a module. Now just add simple python files, like my-tests.py :

1
2
3
4
5
6
7
8
9
10
11
import mymodule
 
class TestAstInfoCli(object):
    def setup(self):
        pass
 
    def teardown(self):
        pass
 
    def test_annuaire_inverse(self):
        assert 1 == 1

As you can see, no need to import anything for doing unittests, not even the standard python unittest module! That’s better than ruby! The downside of this is that nose is an ‘external’ package, so you’ll have to install it first (or set it as a dependency in your setup.py file, as shown above).

If you don’t use a setup.py, you can call nose directly from the command line, with ‘nosetest’.

Now, let’s find an equivalent to the really cool rspec ruby module!

Default behaviour in implementation of STOMP protocol in RabbitMQ with python

4

Why STOMP?

Why STOMP, and not directly AMQP, as I’m using RabbitMQ. No real reason, but the fact that there are less dependencies on a STOMP client, as it’s just a socket with text sent.

Implementations

There are several implementations of the STOMP protocol for Python. The module I chose is python-stomp (version 0.2.9), from Benjamin W. Smith. It’s simple and easy to understand.

Simple Code Examples

sto_send.py:

1
2
3
4
5
6
from stompy.simple import Client
 
stomp = Client(host='rabbitmq2')
stomp.connect(username='guest',password='noneofyourbusiness')
stomp.put('Thomas est une b*te à Tetris...', destination='/queue/jeuvideo')
stomp.disconnect()

sto_receive.py:

1
2
3
4
5
6
7
8
9
10
11
12
from stompy.simple import Client
 
stomp = Client(host='rabbitmq2')
stomp.connect(username='guest',password='noneofyourbusiness')
stomp.subscribe('/queue/jeuvideo')
message = stomp.get()
 
print message.body
 
#stomp.ack(message)
stomp.unsubscribe('/queue/video')
stomp.disconnect()

Everything is working fine, when launching sto_receive.py, I receive the message. But when I launched several receivers, I noticed, that ONLY ONE programs received the message! After some research, I found the answer: As documented in the RabbitMQ wiki, the default exchange is ‘direct’:

[...]when messages leave a queue for a consumer, they are not duplicated. One message, sitting on a queue, is delivered to only one of the available consumers. [...] If there are multiple clients, all SUBSCRIBEing to the same queue, then there will be multiple consumers all on the same queue, leading to round-robin delivery to those clients.

There is an explanation on how to change the behaviour, by changing the exchange type, and some of particular bits (like the id). I even found an example of modification for use in the equivalent STOMP Ruby module.

Here are the modifications. The good news is that there is no need to patch the stompy module, as the author provided the possibility to pass arbitrary parameters to the headers by the use of the ‘conf’ variable.

The important points are:

  • You need to define an exchange of type amq.topic
  • You need to set an id, which is different for each client
  • As you’re using topics, you’ll have to specify a routing_key

sto_receive.py:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
from stompy.simple import Client
import uuid
 
unique_id = uuid.uuid4()
 
stomp = Client(host='rabbitmq2')
stomp.connect(username='guest',password='nonononono')
 
stomp.subscribe('',
                conf={'exchange': 'amq.topic',
                      'routing_key':'x.#',
                      'id': unique_id,
                      })
 
# Wait for a message to appear
while 1:
    message = stomp.get()
    print message.body
 
#stomp.ack(message)
stomp.unsubscribe('',conf={'id': unique_id})
stomp.disconnect()

sto_send.py:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
from stompy.simple import Client
 
 
 
stomp = Client(host='rabbitmq2')
stomp.connect(username='guest',password='nonononono')
 
 
for i in range(10000):
 stomp.put('Thomas est une b*te au Tetris...', destination='x.y',
          conf={'exchange':'amq.topic',
                #  'routing_key':'x.y'
                })
 
stomp.disconnect()

How to dial a number using Asterisk and Python

2

I didn’t find any example in Python on how to Dial a number from an Asterisk server and link it to another channel. So here’s a quick code to do it. You just need to have a manager defined in your /etc/asterisk/manager.conf defined.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
import socket
 
HOST="192.168.1.116"
PORT=5038
 
p = """Action: login
Events: off
Username: %(username)s
Secret: %(password)s
 
Action: originate
Channel: SIP/%(local_user)s
WaitTime: 60
CallerId: 600
Exten: %(phone_to_dial)s
Context: default
Priority: 1
 
Action: Logoff
"""
def click_to_call(phone_to_dial, username, password, local_user):
    pattern = p % {
            'phone_to_dial': phone_to_dial,
            'username': username,
            'password': password,
            'local_user': local_user}
 
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.connect((HOST, PORT))
 
    data = s.recv(1024)
    for l in pattern.split('\n'):
        print "Sending>", l
        s.send(l+'\r\n')
        if l == "":
            data = s.recv(1024)
            print data
    data = s.recv(1024)
    s.close()
 
if __name__ == '__main__':
    click_to_call(phone_to_dial='123456789',
                  username='manager_login', password='yourpass',
                  local_user='600')

Twisted Logo

Making and deploying a Twisted project as a service under Windows

0

What is the goal?

The goal is to implement a program in Python+Twisted (using PB for network access) under Windows XP or 2000+, that can be run before a user logs on, so it has to be a windows service, launched automatically, at boot. Another goal is to show some developement patterns in Twisted. You will find a lot of ‘theoretical’ patterns about how to make singletons/borgs, proxy, and stuff, but I never found patterns about ‘Twisted code’, except for the wonderful ‘finger tutorial’ by the squishy moshez. This tutorial can be split in two parts: The first one is about writing a good skeleton for your Twisted development. The second part is about making a Windows service.

Note that these instructions (the service building part) will not work under Windows 98, 95 or older. (more…)

Go to Top