kenneho.net

Spying on Python functions

February 19, 2019 Kenneth Holter Comments 0 Comment

I was writing test code for a Python project, and needed to spy on the functions of one of the main object’s, called “server”, object instances called “rest_client”. By simply replacing the “rest_client” instance with a mock that passes the actual call to the wrapped object, I’m able to execute the production code of “rest_client” while later on making assertions.

Developing .NET applications without installing .NET SDK

November 29, 2018 Kenneth Holter Comments 0 Comment

I recently developed a small .NET application. I considered installing the .NET SDK on my laptop (on which I were to develop the .NET application), but was unsure of which version I would need, and how I would potentially downgrade the SDK version if needed. Instead of installing the SDK on my laptop, I decided to run the SDK inside a Docker container. This way I could easily switch between SDK versions if needed, as all I’d have to do…

Read More Read More

Docker build fails with “no matching manifest for unknown in the manifest list entries”

November 22, 2018 Kenneth Holter Comments 0 Comment

I recently got this error when trying to build a Windows container (specifically microsoft/dotnet-framework:4.7.2-sdk) on my Windows 10 laptop: no matching manifest for unknown in the manifest list entries I was surprised that I couldn’t build Windows containers on my Windows laptop, but then I recalled that Docker for Windows let you choose between running Linux containers or Windows containers. As I until now had exclusively ran Linux containers, my Docker was set up to run Linux containers, but configuring Docker to…

Read More Read More

Patching Python instance methods

November 10, 2018 Kenneth Holter Comments 0 Comment

I’m working on a project in which one of our services need to publish messages to a RabbitMQ exchange. Consider this code excerpt from the message queue client:

# file: mq_client.py
import pika

class MQ_Client():
    connection = None
    channel = None
    exchange_name="my_exchange"

    def connect(self):
        self.connection = pika.BlockingConnection(pika.ConnectionParameters(host="mq_server")
        self.channel = connection.channel()

    def publish(self, message):

        properties = pika.BasicProperties(content_type='text/json', delivery_mode=1)

        if self.channel.basic_publish(exchange=self.exchange_name, routing_key='', properties=properties, body=message):
            log.info("Successfully published this message to exchange \"" + self.exchange_name + "\": " + message)
        else:
            log.error("Failed to publish message \"" + message + "\" to the exchange \"" + self.exchange_name + "\"!")
            raise Exception("Failed to publish message to the queue.")

# file: mq_client.py

import pika

class MQ_Client():

connection = None

channel = None

exchange_name="my_exchange"

def connect(self):

self.connection = pika.BlockingConnection(pika.ConnectionParameters(host="mq_server")

self.channel = connection.channel()

def publish(self, message):

properties = pika.BasicProperties(content_type='text/json', delivery_mode=1)

if self.channel.basic_publish(exchange=self.exchange_name, routing_key='', properties=properties, body=message):

log.info("Successfully published this message to exchange \"" + self.exchange_name + "\": " + message)

else:

log.error("Failed to publish message \"" + message + "\" to the exchange \"" + self.exchange_name + "\"!")

raise Exception("Failed to publish message to the queue.")

In my unit tests I wanted to verify that an unsuccessful attempt to publish the message, would cause trigger an exception to be thrown. To test this, I needed basic_publish to return False. At first I found it tricky to patch instance methods such as basic_publish , but after coming across this excellent…

Read More Read More

Testing a Flask REST API using Factory Boy and an in-memory sqlite database

September 28, 2018 Kenneth Holter Comments 0 Comment

I’m developing a REST API using Python’s Flask microframework. As I wanted to do test driven development to ensure good test coverage, I needed a working testing framework. After checking out a few different options, I found one that works well for my use case. I thought I’d share my setup in case it may be of help to others. First, this is how I create the server:

# File: server/server.py
import connexion
from database import db
def create_app(config):
    connexion_app = connexion.FlaskApp(__name__)  # For stock Flask users, use something like "app <span class="pl-k">=</span> Flask(<span class="pl-c1">__name__</span>)" instead
    connexion_app.add_api('api.yaml', strict_validation=True)
    app = connexion_app.app
    app.config['SQLALCHEMY_DATABASE_URI'] = config.get_database_uri()
    app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
    db.init_app(app)
    return app

# File: server/server.py

import connexion

from database import db

def create_app(config):

connexion_app = connexion.FlaskApp(__name__) # For stock Flask users, use something like "app <span class="pl-k">=</span> Flask(<span class="pl-c1">__name__</span>)" instead

connexion_app.add_api('api.yaml', strict_validation=True)

app = connexion_app.app

app.config['SQLALCHEMY_DATABASE_URI'] = config.get_database_uri()

app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False

db.init_app(app)

return app

Passing custom HTTP headers using Flask’s test_client

September 22, 2018 Kenneth Holter Comments 0 Comment

Passing custon HTTP headers using Flask’s test_client object should have been as easy as doing this:

headers = { 'API-KEY': 'myKey' }            
response = self.client.get('/users', follow_redirects=True, headers=headers)

1 2	headers = { 'API-KEY': 'myKey' } response = self.client.get('/users', follow_redirects=True, headers=headers)

For some reason this didn’t work in my Flask project. In case others have the same issue as I did, maybe the solution I found to work may work for others too:

import unittest
from server import create_app

class UsersTests(unittest.TestCase):

    def setUp(self):
        app = create_app()
        self.client = app.test_client()

    def test_custom_headers(self):
        self.client.environ_base['HTTP_API_KEY'] = "mySecretKey"
        self.client.environ_base['HTTP_ACCEPT'] = 'application/xml'
        response = self.client.get("/users", follow_redirects=True)

import unittest

from server import create_app

class UsersTests(unittest.TestCase):

def setUp(self):

app = create_app()

self.client = app.test_client()

def test_custom_headers(self):

self.client.environ_base['HTTP_API_KEY'] = "mySecretKey"

self.client.environ_base['HTTP_ACCEPT'] = 'application/xml'

response = self.client.get("/users", follow_redirects=True)

Mocking SQLAlchemy models in Python

September 22, 2018 Kenneth Holter Comments 0 Comment

I’m developing an API using Python’s Flask framework. The API uses SQLAlchemy, which means that database rows are made available as regular Python classes (referred to as database models) in my Python code. One of my models looks something like this:

class View_users(db.Model):
   __tablename__ = 'view_active_users'
   name = db.Column(db.String(80), unique=True, nullable=False)

class View_users(db.Model):

__tablename__ = 'view_active_users'

name = db.Column(db.String(80), unique=True, nullable=False)

My Data Access Object (DAO) for communicating with the database models looks like this:

# File: server/daos/users.py
from server.models.users import View_users
class UsersDAO():
    def get_data(self):
    return View_users.query.all()

# File: server/daos/users.py

from server.models.users import View_users

class UsersDAO():

def get_data(self):

return View_users.query.all()

While writing tests for the API, I wanted to mock out the database models, so that my tests wouldn’t depend on an actual…

Read More Read More

Alarm clock for kids

August 16, 2016 Kenneth Holter Comments 0 Comment

I’ve created a very simple Raspberry Pi project that I use to let my kid know if its daytime or nighttime. When I say goodnight and leave my kids bedroom, I trigger a simple script on my Raspberry Pi with a 7″ touchscreen attached. The screen will display a daytime image for a few minutes, before switching to a nighttime image. The nighttime image is an indication for my kid to put down her toys and go to bed. Over…

Read More Read More

Using the “Oboe” Javascript library to speed up data processing

February 13, 2016 Kenneth Holter Comments 0 Comment

One of our node.js applications at work was designed to call an internal rest service, cache the whole response, and iterate over the data to separate relevant data from non-relevant data, and write the relevant data to file. To speed things up a bit, we wanted to process, filter and write the data to file as it arrived. For this task, the excellent Javascript library “Oboe” came in handy:

var oboe = require('oboe');

var fs = require('fs');

var jsonFileStream = fs.createWriteStream('myfile.json')[...]
var acceptHeader = {
    Accept: 'application/json'
}
var requestHeader = {
    url: "http://rest.example.local/myapp",
    headers: acceptHeader
}
oboe(requestHeader)
    .node('{ref hostname}', function(item) {
        handleItem(node, write_item_to_file);
    })
    .done(function() {
        jsonFileStream.end();
    });

function handleItem(item, callback) {
    if (isInterestingData(item)) {
        callback(item);
    }
}

function isInterestingData(item) {...}

function write_item_to_file(item) {...}

var oboe = require('oboe');

var fs = require('fs');

var jsonFileStream = fs.createWriteStream('myfile.json')[...]

var acceptHeader = {

Accept: 'application/json'

}

var requestHeader = {

url: "http://rest.example.local/myapp",

headers: acceptHeader

}

oboe(requestHeader)

.node('{ref hostname}', function(item) {

handleItem(node, write_item_to_file);

})

.done(function() {

jsonFileStream.end();

});

function handleItem(item, callback) {

if (isInterestingData(item)) {

callback(item);

}

function isInterestingData(item) {...}

function write_item_to_file(item) {...}

Keywords: HTTP response streaming

Using Maven to build and run Java applications

March 10, 2015 Kenneth Holter Comments 0 Comment

Say you want to have Maven handle the dependencies for your java application containing the well known “public static void main” method. Create a Maven project, put your MyApplication.java file under the “src”-folder, and add something like this to you pom.xml:

<?xml version="1.0" encoding="UTF-8"?>
<build>
   <sourceDirectory>src</sourceDirectory>
   <resources>
      <resource>
         <directory>src</directory>
         <excludes>
            <exclude>**/*.java</exclude>
         </excludes>
      </resource>
   </resources>
   <plugins>
      <plugin>
         <artifactId>maven-compiler-plugin</artifactId>
         <version>3.1</version>
         <configuration>
            <source>1.7</source>
            <target>1.7</target>
         </configuration>
      </plugin>
      <plugin>
         <groupId>org.codehaus.mojo</groupId>
         <artifactId>exec-maven-plugin</artifactId>
         <version>1.3.2</version>
         <executions>
            <execution>
               <goals>
                  <goal>java</goal>
               </goals>
            </execution>
         </executions>
         <configuration>
            <includePluginDependencies>true</includePluginDependencies>
            <executable>java</executable>
            <mainClass>MyApplication</mainClass>
         </configuration>
      </plugin>
   </plugins>
</build>

<?xml version="1.0" encoding="UTF-8"?>

<build>

</excludes>

</resource>

</resources>

<artifactId>maven-compiler-plugin</artifactId>

</configuration>

</plugin>

<groupId>org.codehaus.mojo</groupId>

<artifactId>exec-maven-plugin</artifactId>

<goals>

</goals>

</execution>

</executions>

<mainClass>MyApplication</mainClass>

</configuration>

</plugin>

</plugins>

</build>

Now you can run your application using the exec-maven-plugin, like this:

mvn clean compile exec:java -Dexec.args="arg1 arg2 arg3"

1	mvn clean compile exec:java -Dexec.args="arg1 arg2 arg3"