diff options
Diffstat (limited to 'services/database/db_cms.py')
| -rw-r--r-- | services/database/db_cms.py | 265 |
1 files changed, 265 insertions, 0 deletions
diff --git a/services/database/db_cms.py b/services/database/db_cms.py new file mode 100644 index 0000000..3c2b2c6 --- /dev/null +++ b/services/database/db_cms.py @@ -0,0 +1,265 @@ + +# ============LICENSE_START========================================== +# org.onap.vvp/test-engine +# =================================================================== +# Copyright © 2017 AT&T Intellectual Property. All rights reserved. +# =================================================================== +# +# Unless otherwise specified, all software contained herein is licensed +# under the Apache License, Version 2.0 (the “License”); +# you may not use this software except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# +# +# +# Unless otherwise specified, all documentation contained herein is licensed +# under the Creative Commons License, Attribution 4.0 Intl. (the “License”); +# you may not use this documentation except in compliance with the License. +# You may obtain a copy of the License at +# +# https://creativecommons.org/licenses/by/4.0/ +# +# Unless required by applicable law or agreed to in writing, documentation +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# +# ============LICENSE_END============================================ +# +# ECOMP is a trademark and service mark of AT&T Intellectual Property. +import psycopg2 +from wheel.signatures import assertTrue + +from services.constants import Constants +from services.database.db_general import DBGeneral +from services.frontend.base_actions.click import Click +from services.frontend.base_actions.enter import Enter +from services.frontend.base_actions.wait import Wait +from services.frontend.fe_dashboard import FEDashboard +from services.frontend.fe_general import FEGeneral +from services.frontend.fe_user import FEUser +from services.helper import Helper +from services.logging_service import LoggingServiceFactory +from services.session import session + + +logger = LoggingServiceFactory.get_logger() + + +class DBCMS: + + @staticmethod + def insert_query(queryStr): + try: + nativeIceDb = psycopg2.connect( + DBGeneral.return_db_native_connection('cms_db')) + dbConn = nativeIceDb + cur = dbConn.cursor() + logger.debug("Query: " + queryStr) + cur.execute(queryStr) + dbConn.commit() + dbConn.close() + logger.debug("Insert query success!") + # If failed - count the failure and add the error to list of errors. + except: + raise Exception("Couldn't fetch answer using the given query.") + + @staticmethod + def update_query(queryStr): + try: + nativeIceDb = psycopg2.connect( + DBGeneral.return_db_native_connection('cms_db')) + dbConn = nativeIceDb + cur = dbConn.cursor() + logger.debug("Query: " + queryStr) + cur.execute(queryStr) + dbConn.commit() + dbConn.close() + logger.debug("Update query success!") + # If failed - count the failure and add the error to list of errors. + except: + raise Exception("Couldn't fetch answer using the given query.") + + @staticmethod + def select_query(queryStr): + try: + nativeIceDb = psycopg2.connect( + DBGeneral.return_db_native_connection('cms_db')) + dbConn = nativeIceDb + cur = dbConn.cursor() + logger.debug("Query: " + queryStr) + cur.execute(queryStr) + result = str(cur.fetchone()) + if(result.find("',)") != -1): # formatting strings e.g uuid + result = result.partition('\'')[-1].rpartition('\'')[0] + elif(result.find(",)") != -1): # formatting ints e.g id + result = result.partition('(')[-1].rpartition(',')[0] + dbConn.close() + logger.debug("Query result: " + str(result)) + return result + # If failed - count the failure and add the error to list of errors. + except: + raise Exception("Couldn't fetch answer using the given query.") + + @staticmethod + def get_cms_category_id(categoryName): + logger.debug("Get DBCMS category id for name: " + categoryName) + queryStr = "SELECT id FROM public.blog_blogcategory WHERE title = '%s' LIMIT 1;" % ( + categoryName) + logger.debug("Query : " + queryStr) + result = DBCMS.select_query(queryStr) + return result + + @staticmethod + def insert_cms_new_post(title, description, categoryName): + logger.debug("Insert new post : " + title) + queryStr = "INSERT INTO public.blog_blogpost" \ + "(comments_count, keywords_string, rating_count, rating_sum, rating_average, title, slug, _meta_title, description, gen_description, created, updated, status, publish_date, expiry_date, short_url, in_sitemap, content, allow_comments, featured_image, site_id, user_id) "\ + "VALUES (0, '', 0, 0, 0, '%s', '%s-slug', '', '%s', true, current_timestamp - interval '1 day', current_timestamp - interval '2 day', 2, current_timestamp - interval '1 day', NULL, '', true, '<p>%s</p>', true, '', 1, 1);" % ( + title, title, description, description) + logger.debug("Query : " + queryStr) + DBCMS.insert_query(queryStr) + post_id = DBCMS.get_last_added_post_id() + categoryId = DBCMS.get_cms_category_id(categoryName) + DBCMS.add_category_to_post(post_id, categoryId) + return post_id + + @staticmethod + def get_last_added_post_id(): + logger.debug("Get the id of the post inserted") + queryStr = "select MAX(id) FROM public.blog_blogpost;" + logger.debug("Query : " + queryStr) + result = DBCMS.select_query(queryStr) + return result + + @staticmethod + def update_days(xdays, title): + logger.debug("Get the id of the post inserted") +# queryStr = "select MAX(id) FROM public.blog_blogpost;" + queryStr = "UPDATE public.blog_blogpost SET created=current_timestamp - interval '%s day' WHERE title='%s';" % ( + xdays, title) + logger.debug("Query : " + queryStr) + result = DBCMS.update_query(queryStr) + return result + + @staticmethod + def add_category_to_post(postId, categoryId): + logger.debug("bind category into inserted post: " + postId) + queryStr = "INSERT INTO public.blog_blogpost_categories(blogpost_id, blogcategory_id) VALUES (%s, %s);" % ( + postId, categoryId) + logger.debug("Query : " + queryStr) + DBCMS.insert_query(queryStr) + + @staticmethod + def get_documentation_page_id(): + logger.debug("Retrive id of documentation page: ") + queryStr = "SELECT id FROM public.pages_page WHERE title = 'Documentation' LIMIT 1;" + logger.debug("Query : " + queryStr) + result = DBCMS.select_query(queryStr) + return result + + @staticmethod + def get_last_inserted_page_id(): + logger.debug("Retrive id of last page inserted: ") + queryStr = "select MAX(id) FROM public.pages_page;" + logger.debug("Query : " + queryStr) + result = DBCMS.select_query(queryStr) + return result + + @staticmethod + def delete_old_tips_of_the_day(): + logger.debug("Delete all posts ") + queryStr = "DELETE FROM public.blog_blogpost_categories WHERE id>0;" + logger.debug("Query : " + queryStr) + DBCMS.insert_query(queryStr) + queryStr = "DELETE FROM public.blog_blogpost WHERE id>0;;" + logger.debug("Query : " + queryStr) + DBCMS.insert_query(queryStr) + + @staticmethod + def insert_page(title, content, parent_id=None): + logger.debug("Retrive id of documentation page: ") + if parent_id is None: + parent_id = DBCMS.get_documentation_page_id() + queryStr = "INSERT INTO public.pages_page(" \ + "keywords_string, title, slug, _meta_title, description, gen_description, created, updated, status, publish_date, expiry_date, short_url, in_sitemap, _order, in_menus, titles, content_model, login_required, parent_id, site_id)" \ + "VALUES ('', '%s', '%s-slug', '', '%s', true, current_timestamp - interval '1 day', current_timestamp - interval '1 day', 2, current_timestamp - interval '1 day', NULL, '', true, 0, '1,2,3', '%s', 'richtextpage', true, %s, 1);" % ( + title, title, content, title, parent_id) + logger.debug("Query : " + queryStr) + DBCMS.insert_query(queryStr) + + createdPageId = DBCMS.get_last_inserted_page_id() + logger.debug( + "Bind the page with the rich text content related to this page") + queryStr = "INSERT INTO public.pages_richtextpage(page_ptr_id, content) VALUES (%s, '<p>%s</p>');" % ( + createdPageId, content) + logger.debug("Query : " + queryStr) + DBCMS.insert_query(queryStr) + return createdPageId + + @staticmethod + def create_faq(): + title = "title_FAQ" + Helper.rand_string("randomString") + description = "description_FAQ_" + Helper.rand_string("randomString") + DBCMS.delete_old_tips_of_the_day() + postId = DBCMS.insert_cms_new_post(title, description, "FAQ") + assertTrue(len(postId) > 0 and not None) + return title, description + + @staticmethod + def create_news(): + title = "title_News" + Helper.rand_string("randomString") + description = "description_News" + Helper.rand_string("randomString") + postId = DBCMS.insert_cms_new_post(title, description, "News") + assertTrue(len(postId) > 0 and not None) + return title, description + + @staticmethod + def create_announcement(): + title = "title_Announcement_" + Helper.rand_string("randomString") + description = "description_Announcement_" + \ + Helper.rand_string("randomString") + postId = DBCMS.insert_cms_new_post(title, description, "Announcement") + assertTrue(len(postId) > 0 and not None) + return title, description + + @staticmethod + def create_page(parent_id=None): + title = "title_Of_Page_" + Helper.rand_string("randomString") + description = "description_Of_Page_" + \ + Helper.rand_string("randomString") + createdPageId = DBCMS.insert_page(title, description) + assertTrue(len(createdPageId) > 0 and not None) + return title, description + + @staticmethod + def update_X_days_back_post(title, xdays): + logger.debug("Get the id of the post inserted") + queryStr = "UPDATE blog_blogpost SET created = current_timestamp - interval '%s day', publish_date=current_timestamp - interval '%s day' WHERE title= '%s' ;" % ( + xdays, xdays, title) + logger.debug("Query : " + queryStr) + DBCMS.update_query(queryStr) + + @staticmethod + def create_announcements(x): + listOfTitleAnDescriptions = [] + for _ in range(x): + # print x ->str + title = "title_Announcement_" + Helper.rand_string("randomString") + description = "description_Announcement_" + \ + Helper.rand_string("randomString") + postId = DBCMS.insert_cms_new_post( + title, description, "Announcement") + assertTrue(len(postId) > 0 and not None) + xList = [title, description] + listOfTitleAnDescriptions.append(xList) + return listOfTitleAnDescriptions |