DEV Community: Ted

How to read a function's Stdout logs in Go tests

Ted — Sun, 15 Oct 2023 16:30:22 +0000

Being able to read the output of a function can be useful to read some good your tests are failing. Here's how I did it.

The problem

I am currently in the middle of building my lightweight configuration loader called Pluma. I was writing tests for one of the
exported helper functions FromEnv which is supposed to read configurations from the environment variables and load them into a map. All the test cases were passing
except the ones where the user passes a prefix for the variables.

Here a cut-down version of the test file for context:


package pluma

import (
    "bytes"
    "fmt"
    "log"
    "reflect"
    "strings"
    "testing"
)


type mockConfig struct{ opts map[string]string }

func TestFromEnv(t *testing.T) {
    prefix := "TEST_"
    options := make(map[string]string)

    // construct keys and options

    tests := []struct {
        keys   []string
        prefix string
        expect map[string]string
        name   string
    }{
        {
            keys:   keys,
            expect: options,
            name:   "Lowercase keys without prefix, expect full config",
        },
        {
            keys:   keys,
            expect: options,
            prefix: prefix,
            name:   "Lowercase keys with prefix, expect full config",
        },

    }
    for _, test := range tests {
        cfg := mockConfig{opts: make(map[string]string)}
        t.Run(test.name, func(t *testing.T) {
            FromEnv(test.keys, &cfg, test.prefix)
            if !reflect.DeepEqual(cfg.opts, test.expect) {
                t.Errorf("Test Failed, expect %v, have %v", test.expect, cfg.opts)
            }
        })
    }

}

Suppose we have a variable foo defined set to bar

export BUZ=fiz
export TEST_FOO=bar

There were no errors with the first test without a prefix. BUZ is loaded as expected.
If we set a prefix of TEST_ then FOO should be set to bar in the config, but that doesn't happen. Instead, we get this error message.

Test Failed, expect map[FOO:bar] have map[FOO:]

Without knowing the internal state of the function, it is difficult to debug it without running it normally. This is only one test case, i.e. if the function correctly handles a prefix.
In a normal test file, there can be dozens of test cases, it can be very impractical to have to run the function, or the entire program just so we can use the debugger. Adding log statements
can be a bit faster to find the problem and then get rid of them when debugging is over.

The solution

When we run automated tests, only the error messages defined in the tests are printed when a test fails. Any standard output (stdout) messages are ignored. To solve this, we can temporally redirect
the logs to a different place then read from there.

One option is to use a file, but seeing as we are working with (probably) unit tests, we want to keep our requirements as low as possible.
Lucky for us, interfaces are king in Go and any writer that implements io.Writer will do. We will just need to find a way to read from where that writer is writing.
The best writer for this case is the bytes.Buffer. Is both an io.Reader and io.Writer so we can use it to quickly save our tests' output and read it later.

t.Run(test.name, func(t *testing.T) {
    var bf bytes.Buffer
    log.SetOutput(&bf)
    t.Cleanup(func() {
        log.SetOutput(os.Stdout)
    })
    FromEnv(test.keys, &cfg, test.prefix)
    if !reflect.DeepEqual(cfg.opts, test.expect) {
        t.Errorf("Test Failed, expect %v, have %v", test.expect, cfg.opts)
    }
    t.Log(bf.String())
})

In this iteration of the code, we are creating a new buffer with var bf bytes.Buffer and switching the default log output from os.Stdout to our newly created buffer. We are then calling the function being tested, in this case, FromEnv, and checking output validity.

Just like with all resources in Go, we must remember to clean up and close them after, so why not schedule a clean-up so we don't forget?
The anonymous function in t.Cleanup() will be run when all the tests and sub-tests have completed. It will return normal log output to os.Stdout allowing us to read what we stored in our buffer bf by calling t.Log(bf.String()). I don't know about you, but I can't read binary, so we convert the buffer to a string before logging it.

This only works with log statements, not fmt.Print and friends. I will keep looking for a way to show the fmt.Print statements as well, but I think it is better to only have it in logs. We don't want to print debug messages to the client in a production environment. If you know a way to do it with fmt, write a comment or send me an email or dm on twitter.

The bug

So after making this change to the test, I added some log statements and discovered that the lookup key was actually set to TEST__FOO and not TEST_FOO as expected. Turns out I was adding an underscore after the prefix. I prefer if the user explicitly adds the underscore, so I have removed the automatic addition.
I don't know how long it would have taken me to discover the bug. I am also glad I wrote the test because that bug would have gone into production and caused more problems than the Pluma is aiming to solve.

Creating a recursive Django model serializer without Django Rest Framework

Ted — Fri, 27 Jan 2023 20:39:13 +0000

In this article we will look at how to create your own recursive serializer in Django without using the Django Rest Framework
Sometimes when working with Django, we may have some data that we want to serialize (convert to JSON) but we do not have the option of
using the serializers that come with Django Rest Framework. The data can also take the same form for many cases and writing a new serializer
for all of those cases can be tedious and repetitive. This sounds like something a utility function that can be used to generalize the logic and be reused where needed, hence this article's existence.

By the end of the article, we should have a utility function like this:

import json

from somewhere import ModelFoo
from utilities import generic_serializer

instances = ModelFoo.objects.filter(some_filter)
serialized_instances = generic_serializer(instance, i_fields=['bar'], i_models=['baz'])

with open('Foo_data.json', 'w') as fd:
    json.dump(serialized_instances, fd, default=str, indent=2)

Defining the models

To explain things better, let us use an example. Suppose we have a Django application that stores data for a chain of car dealerships.
Each dealership has cars that customers go to one of them to buy a car. The example models.py is shown below:

# models.py
from django.db import models

class Dealership(models.Model):
    name = models.CharField(max_length=20) 
    street_name = models.CharField(max_length=20) 
    zip_code = models.CharField(max_length=6) 

    def __str__(self):
        return self.name

class Car(models.Model):
    name = models.CharField(max_length=20) 
    brand = models.CharField(max_length=20) 
    year = models.CharField(max_length=20) 
    price = models.DecimalField(max_digits=7, decimal_places=2)
    dealership = models.ForeignKey(Dealership, on_delete=models.CASCADE)

    def __str__(self):
        return f"{self.name} {self.brand}"

Creating the generic serializer

Finding the fields

To start us off the first thing our serializer need to be able to do to create a dict of the fields and values of our given model instance. In this article, we will focus on the
Car model instance and try to serialize it. To give us something to work with we create some sample
instances:

# backups.py
dealership = Dealership.objects.create(name="LopezCars", street_name="Tarmac", zip_code="90210")
# {"id":1, "name":"LopezCars", "street_name": "Tarmac", "zip_code": "90210"}

car = Car.objects.create(name="Camión ", brand="Suave", year="2020", price="2283", dealership=1)
# {"id":1, "name":"Camión", "brand":"Suave", "year":"2020", "price":Decimal('2283.00'), "dealership": 1}

The simplest way to convert a model instance to a dict in django is to use the model_to_dict function that is built into django.
To use it, simply import it and pass a model instance to it. It supports include all or exclude all but the specified fields.

>>> from django.forms import model_to_dict
>>> from .backups import car
>>>
>>> model_to_dict(car)
>>> {"name":"Camión", "model":"Suave", "year":"2020", "price":Decimal('2283.00'), "dealership": 1}
>>>
>>> model_to_dict(car, fields=["brand", "price"])
>>> {"model":"Suave", "price":Decimal('2283.00')}
>>>
>>> model_to_dict(car, exclude=["brand", "price"])
>>> {"name":"Camión", "year":"2020", "dealership": 1}

While model_to_dict is very useful for a simple model, it is not powerful enough for what we may need, like recursively fetching related models. Django models have the _meta API which is an instance an django.db.models.options.Options object that allows us to fetch all the field instances of a model. One of the properties available in _meta is the fields which is a django.utils.datastructures.ImmutableList.

We can use this list to get all fields and construct a dict object out of it like so:

# backups.py
...
fields = {}
for f in car._meta.fields:
    fields[f.name] = getattr(model, f.name)

# if you wish to use a comprehension:
fields = {f.name: getattr(model, f.name) for f in car._meta.fields}

# {"id":1, "name":"Camión", "brand":"Suave", "year":"2020", "price":Decimal('2283.00'), "dealership": <Dealership: LopezCars> }

Looking at the output, you may notice that with this method, we preserve the object Primary Key (id) a.k.a pk and the dealership is not just the id. If we do the same thing for dealership:

# backups.py
...
# if you wish to use a comprehension:
fields = {f.name: getattr(model, f.name) for f in dealership._meta.fields}

# {"id":1, "name":"LopezCars", "street_name": "Tarmac", "zip_code": "90210"}

And now we combine the two:

# backups.py
fields = {f.name: getattr(model, f.name) for f in car._meta.fields}
fields['dealership'] = {f.name: getattr(model, f.name) for f in dealership._meta.fields}

# {"id":1, "name":"Camión", "brand":"Suave", "year":"2020", "price":Decimal('2283.00'), "dealership": {"id":1, "name":"LopezCars", "street_name": "Tarmac", "zip_code": "90210"}}

We have successfully converted the model into a dictionary. This is nice, but it is specific to car so to make it generic using the magic of recursion:

# utilities.py
from django.db.models import Model
def get_fields(model: models.Model, fields: dict = None) -> dict:
    if not fields:
        fields = {}
        for f in model._meta.fields:
            fields[f.name] = getattr(model, f.name)

    for name, value in fields.items():
        if not isinstance(value, model.Model):
            # skip non-relational (ForeignKey) fields
            continue
        fields[name] = getattr(value, "pk") if name in ignore_models else get_fields(value)
    return fields

get_fields(car)

# {"id":1, "name":"Camión", "brand":"Suave", "year":"2020", "price":Decimal('2283.00'), "dealership": {"id":1, "name":"LopezCars", "street_name": "Tarmac", "zip_code": "90210"}}

Now we have a generic function to transform any model into a dict object and that can recurse into the other models within it. If that is all you needed, then you can stop here.
We have made a slightly more powerful version of model_to_dict.

Extra Functionality

The get_fields function is fine the way it is and is perfectly usable now. But what if we don't need the to see the brand field or maybe dealership related model? If we want to get information for more than one car we would have to run the function on both of them. To achieve these things we need to expand it with a few things.

# utilities.py
from django.db.models import Model

def generic_serializer(
    instances: list,
    exclude_models: list = None,
    exclude_fields: list = None,
) -> list:
    def get_fields(model: Model = None, fields: dict = None) -> dict:
        if not fields:
            fields = {}
            for f in model._meta.fields:
                if (
                    not f.is_relation
                    and f.name in exclude_fields
                ):
                    continue
                fields[f.name] = getattr(model, f.name)

        for name, value in fields.items():
            if not isinstance(value, Model):
                continue
            # Replace the excluded related model field with their primary key value
            # Instead of storing the `model.__str__()` value, it shows the model.pk
            # if it is in `exclude_models`
            fields[name] = getattr(value, "pk") if name in exclude_models else get_fields(value)
        return fields

    return [get_fields(instance) for instance in instances]

To check if a field f is a relation field i.e models.ForeignKey field, we can use f.is_relation. For the car model the dealership
field is a relational field to Dealership models. Therefor, the value for for dealership.is_relation is True. This allows the exclusion of non-relation field in exclude_field without excluding a relation field with the same name and vice versa. The line if not is instance(value, Model) checks if there there are any relation fields to trigger a recursion.

We are done! (^O^)/ It is a simple function, but it is very useful.

Thanks for taking the time to read this article and I hope it has been useful to you.

How to create a customizable permission class in Django rest framework

Ted — Sun, 17 Jul 2022 00:31:30 +0000

Permissions can be a hustle to deal with when developing an api. Suppose we have a number of api views and endpoints where the access permissions are very similar to one another with only slight variations.

We could create different permission classes with the slight changes that fit the specific endpoint's needs. That works, but that involves a lot of repetition and duplicated code that can be hard to update later down the line.

This is where an abstract or generic permission class comes in handy. We can define template permissions and logic that can be inherited and extended by child permission classes or the modified in the ViewSet class like so:

# views.py

from somewhere import AbstractPermission

class MyViewSet(viewsets.ModelViewSet):
    serializer_class = MySerializer
    permission_classes = (AbstractPermission,)
    deny_action = {
        "CLIENT": ("create",)
    }

Defining our structure

To get started, we can have an example project for CozyCoin Bank with branches in different countries. Each branch has one manager, several clerks, and marketers. CozyCoin customers are rich and can also have multiple bank accounts. Below are our simple models. I have only included fields necessary for explanations.
All the user models inherit from a default user model class were the common fields for all users are defined. For our specific case, we are using the roles field to specify which group the user falls under, but you can use django groups if you wish. I find assigning roles much easier to manage.

# models.py
...
...
class User(AbstractBaseUser):
    class Roles(models.TextChoices):
        MANAGER = "MANAGER", "Manager"
        SUPPORT = "SUPPORT", "Support"
        CLERK = "CLERK", "Clerk"
        CUSTOMER = "CUSTOMER", "Customer"

class BankBranch(models.Model):
    manager = models.ForeignKey(Manager, on_delete=models.CASCADE)

class Customer(User):
    role = models.CharField(max_length=20, default=User.Roles.CUSTOMER) 

class Manager(User):
    role = models.CharField(max_length=20, default=User.Roles.MANAGER) 

class Clerk(User):
    role = models.CharField(max_length=20, default=User.Roles.CLERK) 

class Support(User):
    role = models.CharField(max_length=20, default=User.Roles.SUPPORT) 

class Account(models.Model):
    customer = models.ForeignKey(Customer, on_delete=models.CASCADE)
    bank_branch = models.ForeignKey(Bank, on_delete=models.CASCADE)

class Transactions(models.Model):
    account = models.ForeignKey(Account, on_delete=models.CASCADE)

...
...

Let's assume that the serializers and views have also been setup with a sample viewset looking like this:

# views.py
class SomeModelViewSet(viewsets.ModelViewSet):
    serializer_class = SomeModelSerializer
    permission_classes = (SomePermissionClass,) # this is the important part
    def get_queryset(self):
        # your queryset logic here
        pass

Creating the abstract permission class

A basic permission class

In order to determine if a user has permissions depending on their role, we first have to create an abstract permission class that inherits from rest framework's BasePermission.
We will deny any anonymous request since we require the user to be logged in to view their role. If you don't have an account of any sort with CozyCoin, then you shouldn't have access to anything protected by these permissions.

# permissions.py
from rest_framework import permissions

class IsRole(permissions.BasePermission):
    def has_permission(self, request, view):
        if request.user.is_anonymous:
            return False 
    def has_object_permission(self, request, view, obj):
        pass

For our permissions to be effective, we first have to determine if the user making the request is in the desired group.
The role property is initialized to None and is supposed to be modified in the child classes. If it is not, then SomeErroris raised.
SomeError being an error of your choosing. This will ensure that the permissions don't silently fail and deny all requests since the role will always be None and therefor not match any roles.

# permissions.py
class IsRole(permissions.BasePermission):
    role = None
    def has_permission(self, request, view):
        if request.user.is_anonymous:
            return False 
        return self.has_role(request):

    def has_object_permission(self, request, view, obj):
        return self.has_role(request)

    def has_role(self, request) -> bool:
        if self.role == None:
            raise SomeError
        return request.user.role == self.role

And that's it! We can create child classes that specify the roles we want to grant access to. For example:

# views.py
from some_location.permissions import IsRole

class IsCustomer(IsRole):
    role = "CUSTOMER"

class CustomerOnlyView(viewsets.ViewSet):
    permission_classes = (IsCustomer,)
    ...

Granting only the owner of an object permissions

The permissions we have defined above are perfectly usable, but if you are keen you might have noticed a problem. Taking the IsCustomer permissions for example, if the request user is not a customer they will be blocked just fine. The problem is the customer can request for any customer data that is not theirs and it will be granted to them. This means that any customer can view any other customer's data in a view protected by these permissions. To help with this issue, we need to tweak our has_object_permission method.

# permissions.py
    ...
    ...
    def has_object_permission(self, request, view, obj):
        owner = getattr(obj, self.role.lower())
        return owner.id == request.user.id

With this small tweak, instead of just giving any data to anyone so long as they have a certain role, we will be able to only limit the
data to what they own or is linked to them. This assumes that the foreign key field pointing to the user table is named the same as the role but in all lowercase. This means that if the role is CLIENT, then the foreign key field in the model is called client. An alternative it to get the name of the fields to check at runtime instead of relying on the role name which could be different.

# permissions
class IsRole(permissions.BasePermission):
    owner_field = None
    ...
    def has_object_permission(self, request, view, obj):
        user = getattr(obj, self.owner_field or self.role.lower()))
        return user.id == request.user.id

class IsCustomer(IsRole):
    role = "CUSTOMER"
    owner_field = "client"

Granting privileged users access

Wait, what if the person asking for the data has higher access rights like an admin or us? We don't want to make the managers feel less powerful. We have different ways we can deal with this problem. How you approach it will depend on what you named your abstract class and how abstract you want to go. The first approach is to use a variable that grants a set of user roles access. In this example, we can name the field allow_staff.

class IsRole(permissions.BasePermission):
    allow_staff = False
    staff_roles = ["ADMIN", "MANAGER"]
    ...
    def has_permission(self, request, view, obj):
        ...
        if self.allow_staff and request.user.role in staff_roles:
            return True
        ...
    def has_object_permission(self, request, view, obj):
        if self.allow_staff and request.user.role in staff_roles:
            return True
        ...

With this approach, we can define the staff_roles and set allow_staff to true to give the people with staff_roles access. If we want to go even more abstract then we can use the second approach were we can define allow_roles instead of staff_roles. With this we don't need to check if allow_staff is true or not, we just check if the current user role is one of the special ones.

Grant access by view action

Now, that we can filter permissions by roles, we can also try filtering by the action being performed. For example we might not want a support staff member should not be able to view all the customers and their data in one place at once. (This is only useful with a non-incremental system object ID. There's nothing stopping Joe from going up the number of client IDs from one in an incremental system). We might want them to only have access to the specific client asking for support and only for a short period of
time or after they have been per-approved by a client issue ticket handling system. Also when Bob from marketing wants to send an email to all customers over 35, who look like they could use a loan, he will need not any permissions to modify customer data. So we will give him and his team read only access.

To achieve this, we can add it another attribute deny_actions to the abstract class. This attribute can be a list containing actions we want to deny on that endpoint. We can then check for the list in the has_permission method:

# permissions.py
    ...
    if self.deny_actions and view.action not in self.deny_actions:
        return True

This solution is simple, but it will also deny any special privileged users from performing the actions. We can change it up slightly by making deny_actions accept a dictionary with the user roles as the keys and denied roles as the values. e.g:

deny_actions = {
    "MARKETING": ["create", "update", "delete"]
}

We can put the logic for that in a separate class method and only call it when we need to. We also raise an exception in case the values are not formatted correctly.

# permissions.py
    ...
    ...
    def is_action_denied(self, request, view):
        if isinstance(self.deny_actions, dict):
            actions = self.deny_actions.get(request.user.role)
        elif isinstance(self.deny_actions, list | tuple):
            actions = self.deny_actions
        else:
            raise TypeError(
                f"Invalid type {type(self.deny_actions)} for deny_actions. Expected iterable"
            )
        if actions and view.action in actions:
            return False
        return True

Getting values from the view

Sometimes you do not want to create an entire class just so you can change 1 value and use it on one view only. What would be point of doing all this work? We just need to have a bit more code and logic.

# permissions.py
    deny_actions = None
    allow_roles = None
    allow_staff = None
    def has_permission(self, request, view):
        deny_actions = getattr(view, 'deny_actions', None)
        allow_roles = getattr(view, 'allow_roles', None)
        allow_staff = getattr(view, 'allow_staff', None)

This will allow us to specify the extra values when creating the child permission classes or within the view like so:

# views.py
class SomeModelViewSet(viewsets.ModelViewSet):
    serializer_class = SomeModelSerializer
    permission_classes = (IsClient,)
    allow_staff = True
    deny_actions = {
        "MANAGER": ["create", "update", "delete"]
    }
    def get_queryset(self):
        pass

The part where we fetch the filters can also bring up a problem if you have many of them, or if we increase the filters in the future.
So we define a method that is called by has_permissions method before doing anything.

# permissions.py
    ...
    def get_or_set_attributes(self, request, view):
    for attribute in dir(self):
        if not attribute.startswith("_"): # ignore any magic methods and private variables
            value = getattr(self, attribute)
            if value == None:
                setattr(self, attribute, getattr(view, attribute, None))

Limitations

So this method of creating things an abstract permissions class is very useful. It means we only have to make a few changes to existing and that's it.
If we need new very specific permissions, we don't have to duplicate some code from somewhere and only change a few things to fit the view. There are some issues with this approach.

Too much code in one place

As we are trying to make the abstract class as versatile and flexible as possible, we have to cram so much logic and checks in it. This can very quickly
become a nightmare to maintain or to read back in 6 months.
Long setup time

It took me quite a while to come up with the class and logic. It works well for the most part, but it might be hard to adapt it to a different application.
Is it worth the time it took to make it? ¯_(ツ)_/¯ Maybe it will be in the long run as I would just copy paste the class into other projects, and instantly
get customizable permissions without defining them once again.
Reliance on role

If you were to remove the role attribute, major parts of the class will have to be redone or rethought. With this structure, a user can only belong to one role.
This limitation could be overcome by allowing the primary search field to be set in child class or in the viewset.
Limited extensibility

There is only so much you can add before it becomes a too much. The best thing at that point would be to create a separate abstract permission class.
This only works class based views.
I have not created or tested a function based view version of this code, so it might not work for that use case

Conclusion

Thanks for getting to the end. Whether you got here by reading or scrolling or copy pasting. You can find the full permission class on github gits