<commit_before>from pytest import fixture, mark
from unittest.mock import Mock

import requests


@mark.usefixtures('integration')
class TestUpdateElasticsearchPolicycompass:

    @fixture
    def registry(self, config):
        return config.registry


    def make_external_resource(self, name, registry, pool_with_catalogs):
        from adhocracy_core import resources, sheets

        context = pool_with_catalogs
        er_appstructs = {
            sheets.name.IName.__identifier__: {'name': name}}
        external_resource = registry.content.create(
            resources.external_resource.IExternalResource.__identifier__,
            parent=context,
            appstructs=er_appstructs)
        return context

    @fixture
    def context_r1(self, registry, pool_with_catalogs):
        return self.make_external_resource('dataset_1',
                                           registry,
                                           pool_with_catalogs)

    @fixture
    def context_r2(self, registry, pool_with_catalogs):
        return self.make_external_resource('dataset_2',
                                           registry,
                                           pool_with_catalogs)

    def make_comment(self, resource_name, registry, context):
        from adhocracy_core import resources, sheets

        comment = registry.content.create(
            resources.comment.IComment.__identifier__,
            parent=context[resource_name])
        registry.content.create(
            resources.comment.ICommentVersion.__identifier__,
            parent=comment,
            appstructs={
                sheets.comment.IComment.__identifier__: {
                    'content': 'This is a comment',
                    'refers_to': context[resource_name]
                }
            })
        return comment

    def delete_comment(self, comment, registry, context):
        from adhocracy_core import utils, resources, sheets

        metadata = utils.get_sheet(comment, sheets.metadata.IMetadata, registry)
        metadata.set({ 'hidden': True })

    def requests_post(self, status_code=200):
        response = requests.Response
        response.status_code = status_code
        post = Mock(name='requests.post', return_value=response)
        return(post)

    def test_elastic_search_update_on_create(self, registry, context_r1, monkeypatch):
        requests_post = self.requests_post()
        monkeypatch.setattr(requests, 'post', requests_post)
        self.make_comment('dataset_1', registry, context_r1)

        requests_post.assert_called_with(
            'http://localhost:9000/policycompass_search/dataset/1/_update',
            json={'doc': {'comment_count': 1}})

    def test_elastic_search_update_on_delete(self, registry, context_r2, monkeypatch):
        requests_post = self.requests_post()
        monkeypatch.setattr(requests, 'post', requests_post)
        comment = self.make_comment('dataset_2', registry, context_r2)

        self.delete_comment(comment, registry, context_r2)
        requests_post.assert_called_with(
            'http://localhost:9000/policycompass_search/dataset/2/_update',
            json={'doc': {'comment_count': 0}})<commit_msg>Extend pcompass subscriber tests to error cases<commit_after>from pytest import fixture, mark, raises
from unittest.mock import Mock

import requests


@mark.usefixtures('integration')
class TestUpdateElasticsearchPolicycompass:

    @fixture
    def registry(self, config):
        return config.registry


    def make_external_resource(self, name, registry, pool_with_catalogs):
        from adhocracy_core import resources, sheets

        context = pool_with_catalogs
        er_appstructs = {
            sheets.name.IName.__identifier__: {'name': name}}
        external_resource = registry.content.create(
            resources.external_resource.IExternalResource.__identifier__,
            parent=context,
            appstructs=er_appstructs)
        return context

    @fixture
    def context_r1(self, registry, pool_with_catalogs):
        return self.make_external_resource('dataset_1',
                                           registry,
                                           pool_with_catalogs)

    @fixture
    def context_r2(self, registry, pool_with_catalogs):
        return self.make_external_resource('dataset_2',
                                           registry,
                                           pool_with_catalogs)

    def make_comment(self, resource_name, registry, context):
        from adhocracy_core import resources, sheets

        comment = registry.content.create(
            resources.comment.IComment.__identifier__,
            parent=context[resource_name])
        registry.content.create(
            resources.comment.ICommentVersion.__identifier__,
            parent=comment,
            appstructs={
                sheets.comment.IComment.__identifier__: {
                    'content': 'This is a comment',
                    'refers_to': context[resource_name]
                }
            })
        return comment

    def delete_comment(self, comment, registry, context):
        from adhocracy_core import utils, resources, sheets

        metadata = utils.get_sheet(comment, sheets.metadata.IMetadata, registry)
        metadata.set({ 'hidden': True })

    def requests_post(self, status_code=200):
        response = requests.Response
        response.status_code = status_code
        post = Mock(name='requests.post', return_value=response)
        return(post)

    def test_elastic_search_update_on_create(self, registry, context_r1, monkeypatch):
        requests_post = self.requests_post()
        monkeypatch.setattr(requests, 'post', requests_post)
        self.make_comment('dataset_1', registry, context_r1)

        requests_post.assert_called_with(
            'http://localhost:9000/policycompass_search/dataset/1/_update',
            json={'doc': {'comment_count': 1}})

    def test_elastic_search_update_on_delete(self, registry, context_r2, monkeypatch):
        requests_post = self.requests_post()
        monkeypatch.setattr(requests, 'post', requests_post)
        comment = self.make_comment('dataset_2', registry, context_r2)

        self.delete_comment(comment, registry, context_r2)
        requests_post.assert_called_with(
            'http://localhost:9000/policycompass_search/dataset/2/_update',
            json={'doc': {'comment_count': 0}})

    def test_elastic_search_update_not_found(self, registry, context_r1, monkeypatch):
        requests_post = self.requests_post(404)
        monkeypatch.setattr(requests, 'post', requests_post)
        self.make_comment('dataset_1', registry, context_r1)

    def test_elastic_search_update_error(self, registry, context_r1, monkeypatch):
        requests_post = self.requests_post(400)
        monkeypatch.setattr(requests, 'post', requests_post)
        with raises(ValueError):
            self.make_comment('dataset_1', registry, context_r1)
{'fraction_non_alphanumeric': 0.08519732155577718, 'fraction_numerical': 0.009118107992591537, 'mean_word_length': 2.559837728194726, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 16, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '259641', 'n_tokens_mistral': 1882, 'n_tokens_neox': 1808, 'n_words': 350}