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.
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…
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…
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:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 |
# 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…
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:
|
1 2 3 4 5 6 7 8 9 10 11 |
# 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 custon HTTP headers using Flask’s test_client object should have been as easy as doing this:
|
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:
|
1 2 3 4 5 6 7 8 9 10 11 12 13 |
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) |
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:
|
1 2 3 |
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:
|
1 2 3 4 5 |
# 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…
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…
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:
|
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 |
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
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:
|
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 |
<?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> |
Now you can run your application using the exec-maven-plugin, like this:
|
1 |
mvn clean compile exec:java -Dexec.args="arg1 arg2 arg3" |