Policy Testing

Because CircleCI’s policy engine uses OPA’s rego, we can write test cases and assettions against it.

Pytest Approach

pytest has great parsing and output utilities, and python is great for dealing with yaml and json.

Concept

A pytest based helper class allows the tuple of circleci config, circleci policy, and circleci metadata to be be processed, and then evaluate the outcome.

Code Samples

sample_test.py

This test class runs 4 tests against our context protection policy file.

Good config passes (single filter, main branch only)
Missing filters fails
Multiple filters fails
Single, non-main filter fails

 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
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66


import pytest
from ..helper import PolicyBundleTest


my_config = 'tests/data/config_boa_with_deploys.yml'
my_meta = 'tests/data/meta_boa.json'


def test_valid_config_ok():
    # Given:  a 'valid' CircleCI Configuration
    circleci = PolicyBundleTest('applied_policies', my_config, my_meta)
    config = circleci.config
    # confirm our config is read (optional)
    assert len(config['workflows']) == 1

    # When: we assess policy
    result = circleci.decision()

    # then is PASS
    assert result['status'] == "PASS" 


'''
This test modifies a valid job to attempt invalid access to a context
'''
def test_unfiltered_access_fails():
    # Given:  a 'valid' CircleCI Configuration
    circleci = PolicyBundleTest('applied_policies', my_config, my_meta)

    # And a modification attempting to access prod context on unfiltered branch 
    circleci.config['workflows']['main']['jobs'][4]['deploy']['context']=["prod-context"]
    # When: we assess policy
    result = circleci.decision()

    # THen is fails
    assert result['status'] == "HARD_FAIL" 



def test_multiple_filters_not_allowed():
    # Given:  a 'valid' CircleCI Configuration
    circleci = PolicyBundleTest('applied_policies', my_config, my_meta)

    # And a modification attempting to add non-main branch to the filter 
    circleci.config['workflows']['main']['jobs'][5]['deploy']['filters']['branches']['only'].append('foo')
    assert len(circleci.config['workflows']['main']['jobs'][5]['deploy']['filters']['branches']['only']) == 2
   
    # When: we assess policy
    result = circleci.decision()

    # THen is fails
    assert result['status'] == "HARD_FAIL" 

def test_single_filter_main_only():
    # Given:  a 'valid' CircleCI Configuration
    circleci = PolicyBundleTest('applied_policies', my_config, my_meta)

    # And a modification attempting to add non-main branch to the filter 
    circleci.config['workflows']['main']['jobs'][5]['deploy']['filters']['branches']['only'][0] = 'mine' #not main
    assert len(circleci.config['workflows']['main']['jobs'][5]['deploy']['filters']['branches']['only']) == 1 # only 1
   
    # When: we assess policy
    result = circleci.decision()

    # THen is fails
    assert result['status'] == "HARD_FAIL" 

Helper.py

 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


import os
import yaml
import json
import subprocess


MINIMAL_CONFIG = {'version': 2.1, 'jobs': {'build': {'docker': [{'image': 'cimg/base:edge'}], 'steps': ['checkout', {'run': 'echo "this is the build job"'}]}}}
MINIMAL_META = {"project_id":"123"}
class PolicyBundleTest:

    def __init__(self, bundle, config=None, meta=None):
        self.bundle = bundle
        self.meta = MINIMAL_META
        if meta:
            with open(meta, "r") as file:
                self.meta = json.load(file)  
        self.config = MINIMAL_CONFIG
        if config:
            with open(config, "r") as file:
                self.config = yaml.safe_load(file)



    def decision(self):
        with open('/tmp/meta.json','w') as metafile:
            json.dump(self.meta,metafile)
        with open('/tmp/config.yml','w') as configfile:
            yaml.dump(self.config,configfile)
        result = subprocess.check_output(["circleci", "policy", "decide"
                                        , "--input", "/tmp/config.yml"
                                        , "--metafile", "/tmp/meta.json"
                                        , self.bundle]
        , input=str(self.config)
        ,stderr=subprocess.STDOUT
        ,universal_newlines=True)   
        
        print(str(result))
        return json.loads(result)
        