<commit_before>try:
    from builtins import str, range
except ImportError:
    pass
import os
import tempfile
from flask import redirect, url_for, current_app
from flask.ext.login import LoginManager, login_user, logout_user, current_user
from sqlalchemy import create_engine, MetaData
from flask_blogging.sqlastorage import SQLAStorage
from flask_blogging import BloggingEngine
from test import FlaskBloggingTestCase
from flask_login import UserMixin
import re
from flask.ext.principal import identity_changed, Identity, \
    AnonymousIdentity, identity_loaded, RoleNeed, UserNeed


class TestUser(UserMixin):
    def __init__(self, user_id):
        self.id = user_id


class TestViews(FlaskBloggingTestCase):

    def _create_storage(self):
        temp_dir = tempfile.gettempdir()
        self._dbfile = os.path.join(temp_dir, "temp.db")
        conn_string = 'sqlite:///'+self._dbfile
        engine = create_engine(conn_string)
        meta = MetaData()
        self.storage = SQLAStorage(engine, metadata=meta)
        meta.create_all(bind=engine)

    def setUp(self):
        FlaskBloggingTestCase.setUp(self)
        self._create_storage()
        self.app.config["BLOGGING_URL_PREFIX"] = "/blog"
        self.engine = BloggingEngine(self.app, self.storage)
        self.login_manager = LoginManager(self.app)

        @self.login_manager.user_loader
        @self.engine.user_loader
        def load_user(user_id):
            return TestUser(user_id)

        @self.app.route("/login/<username>/", methods=["POST"],
                        defaults={"blogger": 0})
        @self.app.route("/login/<username>/<int:blogger>/", methods=["POST"])
        def login(username, blogger):
            this_user = TestUser(username)
            login_user(this_user)
            if blogger:
                identity_changed.send(current_app._get_current_object(),
                                      identity=Identity(username))
            return redirect("/")

        @self.app.route("/logout/")
        def logout():
            logout_user()
            identity_changed.send(current_app._get_current_object(),
                                  identity=AnonymousIdentity())
            return redirect("/")

        for i in range(20):
            tags = ["hello"] if i < 10 else ["world"]
            user = "testuser" if i < 10 else "newuser"
            self.storage.save_post(title="Sample Title%d" % i,
                                   text="Sample Text%d" % i,
                                   user_id=user, tags=tags)

    def tearDown(self):
        os.remove(self._dbfile)

    def test_index(self):
        response = self.client.get("/blog/")
        self.assertEqual(response.status_code, 200)

        response = self.client.get("/blog")
        self.assertEqual(response.status_code, 301)

        response = self.client.get("/blog/5/")
        self.assertEqual(response.status_code, 200)

        response = self.client.get("/blog/5/2/")
        self.assertEqual(response.status_code, 200)

    def test_post_by_id(self):
        response = self.client.get("/blog/page/1/")
        self.assertEqual(response.status_code, 200)

        response = self.client.get("/blog/page/1/sample-title/")
        self.assertEqual(response.status_code, 200)

        response = self.client.get("/blog/page/1")  # trailing slash redirect
        self.assertEqual(response.status_code, 301)

    def test_post_by_tag(self):
        response = self.client.get("/blog/tag/hello/")
        self.assertEqual(response.status_code, 200)

        response = self.client.get("/blog/tag/hello/5/")
        self.assertEqual(response.status_code, 200)

        response = self.client.get("/blog/tag/hello/5/2/")
        self.assertEqual(response.status_code, 200)

    def test_post_by_author(self):
        response = self.client.get("/blog/author/newuser/")
        self.assertEqual(response.status_code, 200)

        response = self.client.get("/blog/author/newuser/5/")
        self.assertEqual(response.status_code, 200)

        response = self.client.get("/blog/author/newuser/5/2/")
        self.assertEqual(response.status_code, 200)

        response = self.client.get("/blog/author/nonexistent_user/")
        assert "No posts found for this user!" in str(response.data)

    def test_editor_get(self):
        user_id = "testuser"
        with self.client:
            # access to editor should be forbidden before login
            response = self.client.get("/blog/editor/")
            self.assertEqual(response.status_code, 401)

            response = self.client.get("/blog/editor/1/")
            self.assertEqual(response.status_code, 401)

            self.login(user_id)
            self.assertEquals(current_user.get_id(), user_id)
            response = self.client.get("/blog/editor/")
            assert response.status_code == 200

            for i in range(1, 21):
                # logged in user can edit their post, and will be redirected
                # if they try to edit other's post
                response = self.client.get("/blog/editor/%d/" % i)
                expected_status_code = 200 if i <= 10 else 302
                self.assertEqual(response.status_code, expected_status_code,
                                 "Error for item %d %d" %
                                 (i, response.status_code))
            # logout and the access should be gone again
            self.logout()
            response = self.client.get("/blog/editor/")
            self.assertEqual(response.status_code, 401)

            response = self.client.get("/blog/editor/1/")
            self.assertEqual(response.status_code, 401)

    def test_editor_post(self):
        user_id = "testuser"
        with self.client:
            # access to editor should be forbidden before login
            response = self.client.get("/blog/page/21/",
                                       follow_redirects=True)
            assert "The page you are trying to access is not valid!" in \
                   str(response.data)

            response = self.client.post("/blog/editor/")
            self.assertEqual(response.status_code, 401)

            response = self.client.post("/blog/editor/1/")
            self.assertEqual(response.status_code, 401)

            self.login(user_id)
            self.assertEquals(current_user.get_id(), user_id)

            response = self.client.post("/blog/editor/",
                                        data=dict(text="Test Text",
                                                  tags="tag1, tag2"))
            # should give back the editor page
            self.assertEqual(response.status_code, 200)

            response = self.client.post("/blog/editor/",
                                        data=dict(title="Test Title",
                                                  text="Test Text",
                                                  tags="tag1, tag2"))
            self.assertEqual(response.status_code, 302)

            response = self.client.get("/blog/page/21/")
            self.assertEqual(response.status_code, 200)

    def test_editor_edit_page(self):
        user_id = "testuser"
        with self.client:
            self.login(user_id)
            response = self.client.post("/blog/editor/1/",
                                        data=dict(
                                            title="Sample Title0-Edited",
                                             text="Sample Text0-Edited",
                                             tags="tag1, tag2"
                                        ))

            response = self.client.get("/blog/100/")
            self.assertEqual(response.status_code, 200)
<commit_msg>Fix PEP8.<commit_after>try:
    from builtins import str, range
except ImportError:
    pass
import os
import tempfile
from flask import redirect, url_for, current_app
from flask.ext.login import LoginManager, login_user, logout_user, current_user
from sqlalchemy import create_engine, MetaData
from flask_blogging.sqlastorage import SQLAStorage
from flask_blogging import BloggingEngine
from test import FlaskBloggingTestCase
from flask_login import UserMixin
import re
from flask.ext.principal import identity_changed, Identity, \
    AnonymousIdentity, identity_loaded, RoleNeed, UserNeed


class TestUser(UserMixin):
    def __init__(self, user_id):
        self.id = user_id


class TestViews(FlaskBloggingTestCase):

    def _create_storage(self):
        temp_dir = tempfile.gettempdir()
        self._dbfile = os.path.join(temp_dir, "temp.db")
        conn_string = 'sqlite:///'+self._dbfile
        engine = create_engine(conn_string)
        meta = MetaData()
        self.storage = SQLAStorage(engine, metadata=meta)
        meta.create_all(bind=engine)

    def setUp(self):
        FlaskBloggingTestCase.setUp(self)
        self._create_storage()
        self.app.config["BLOGGING_URL_PREFIX"] = "/blog"
        self.engine = BloggingEngine(self.app, self.storage)
        self.login_manager = LoginManager(self.app)

        @self.login_manager.user_loader
        @self.engine.user_loader
        def load_user(user_id):
            return TestUser(user_id)

        @self.app.route("/login/<username>/", methods=["POST"],
                        defaults={"blogger": 0})
        @self.app.route("/login/<username>/<int:blogger>/", methods=["POST"])
        def login(username, blogger):
            this_user = TestUser(username)
            login_user(this_user)
            if blogger:
                identity_changed.send(current_app._get_current_object(),
                                      identity=Identity(username))
            return redirect("/")

        @self.app.route("/logout/")
        def logout():
            logout_user()
            identity_changed.send(current_app._get_current_object(),
                                  identity=AnonymousIdentity())
            return redirect("/")

        for i in range(20):
            tags = ["hello"] if i < 10 else ["world"]
            user = "testuser" if i < 10 else "newuser"
            self.storage.save_post(title="Sample Title%d" % i,
                                   text="Sample Text%d" % i,
                                   user_id=user, tags=tags)

    def tearDown(self):
        os.remove(self._dbfile)

    def test_index(self):
        response = self.client.get("/blog/")
        self.assertEqual(response.status_code, 200)

        response = self.client.get("/blog")
        self.assertEqual(response.status_code, 301)

        response = self.client.get("/blog/5/")
        self.assertEqual(response.status_code, 200)

        response = self.client.get("/blog/5/2/")
        self.assertEqual(response.status_code, 200)

    def test_post_by_id(self):
        response = self.client.get("/blog/page/1/")
        self.assertEqual(response.status_code, 200)

        response = self.client.get("/blog/page/1/sample-title/")
        self.assertEqual(response.status_code, 200)

        response = self.client.get("/blog/page/1")  # trailing slash redirect
        self.assertEqual(response.status_code, 301)

    def test_post_by_tag(self):
        response = self.client.get("/blog/tag/hello/")
        self.assertEqual(response.status_code, 200)

        response = self.client.get("/blog/tag/hello/5/")
        self.assertEqual(response.status_code, 200)

        response = self.client.get("/blog/tag/hello/5/2/")
        self.assertEqual(response.status_code, 200)

    def test_post_by_author(self):
        response = self.client.get("/blog/author/newuser/")
        self.assertEqual(response.status_code, 200)

        response = self.client.get("/blog/author/newuser/5/")
        self.assertEqual(response.status_code, 200)

        response = self.client.get("/blog/author/newuser/5/2/")
        self.assertEqual(response.status_code, 200)

        response = self.client.get("/blog/author/nonexistent_user/")
        assert "No posts found for this user!" in str(response.data)

    def test_editor_get(self):
        user_id = "testuser"
        with self.client:
            # access to editor should be forbidden before login
            response = self.client.get("/blog/editor/")
            self.assertEqual(response.status_code, 401)

            response = self.client.get("/blog/editor/1/")
            self.assertEqual(response.status_code, 401)

            self.login(user_id)
            self.assertEquals(current_user.get_id(), user_id)
            response = self.client.get("/blog/editor/")
            assert response.status_code == 200

            for i in range(1, 21):
                # logged in user can edit their post, and will be redirected
                # if they try to edit other's post
                response = self.client.get("/blog/editor/%d/" % i)
                expected_status_code = 200 if i <= 10 else 302
                self.assertEqual(response.status_code, expected_status_code,
                                 "Error for item %d %d" %
                                 (i, response.status_code))
            # logout and the access should be gone again
            self.logout()
            response = self.client.get("/blog/editor/")
            self.assertEqual(response.status_code, 401)

            response = self.client.get("/blog/editor/1/")
            self.assertEqual(response.status_code, 401)

    def test_editor_post(self):
        user_id = "testuser"
        with self.client:
            # access to editor should be forbidden before login
            response = self.client.get("/blog/page/21/",
                                       follow_redirects=True)
            assert "The page you are trying to access is not valid!" in \
                   str(response.data)

            response = self.client.post("/blog/editor/")
            self.assertEqual(response.status_code, 401)

            response = self.client.post("/blog/editor/1/")
            self.assertEqual(response.status_code, 401)

            self.login(user_id)
            self.assertEquals(current_user.get_id(), user_id)

            response = self.client.post("/blog/editor/",
                                        data=dict(text="Test Text",
                                                  tags="tag1, tag2"))
            # should give back the editor page
            self.assertEqual(response.status_code, 200)

            response = self.client.post("/blog/editor/",
                                        data=dict(title="Test Title",
                                                  text="Test Text",
                                                  tags="tag1, tag2"))
            self.assertEqual(response.status_code, 302)

            response = self.client.get("/blog/page/21/")
            self.assertEqual(response.status_code, 200)

    def test_editor_edit_page(self):
        user_id = "testuser"
        with self.client:
            self.login(user_id)
            response = self.client.post(
                "/blog/editor/1/",
                data=dict(title="Sample Title0-Edited",
                          text="Sample Text0-Edited", tags="tag1, tag2"))

            response = self.client.get("/blog/100/")
            self.assertEqual(response.status_code, 200)

{'fraction_non_alphanumeric': 0.11020889267238557, 'fraction_numerical': 0.015912513915264225, 'mean_word_length': 1.908952380952381, 'pattern_counts': {'":': 2, '<': 15, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 38, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6467705', 'n_tokens_mistral': 4350, 'n_tokens_neox': 4033, 'n_words': 817}